NRG (2.11): Fix term handling in candidate state and use higher term from vote request (#5671)

A candidate could incorrectly revert to an older term without resetting
if an old AE arrived with a term that is at least newer than the pterm
but not necessarily newer than the term.

Additionally, we also weren't handling the case that the rest of the
cluster should assume the higher term number from the vote requests once
an isolated candidate node rejoins the cluster.

This PR also rewrites `TestNRGCandidateStepsDownAfterAE` and replaces it
with `TestNRGAssumeHighTermAfterCandidateIsolation` as the old test was
checking for flawed behaviour: it was relying on the term being able to
go backwards after a period of isolation, which is not correct. Instead
we should prove that the rest of the cluster assumes the higher term
from the isolated node.

Co-authored-by: Reuben Ninan <[email protected]>
Signed-off-by: Neil Twigg <[email protected]>

Author: derekcollison

server/raft.go

@@ -3201,11 +3201,11 @@ func (n *raft) processAppendEntry(ae *appendEntry, sub *subscription) {
 			// If the append entry term is newer than the current term, erase our
 			// vote.
 			if ae.term > n.term {
+				n.term = ae.term
 				n.vote = noVote
+				n.writeTermVote()
 			}
 			n.debug("Received append entry in candidate state from %q, converting to follower", ae.leader)
-			n.term = ae.term
-			n.writeTermVote()
 			n.stepdown.push(ae.leader)
 		}
 	}
@@ -3978,8 +3978,8 @@ func (n *raft) processVoteRequest(vr *voteRequest) error {
 			n.debug("Stepping down from %s, detected higher term: %d vs %d",
 				strings.ToLower(n.State().String()), vr.term, n.term)
 			n.stepdown.push(noLeader)
-			n.term = vr.term
 		}
+		n.term = vr.term
 		n.vote = noVote
 		n.writeTermVote()
 	}

server/raft_test.go

@@ -477,7 +477,7 @@ func TestNRGInvalidTAVDoesntPanic(t *testing.T) {
 	c.waitOnAllCurrent()
 }
 
-func TestNRGCandidateStepsDownAfterAE(t *testing.T) {
+func TestNRGAssumeHighTermAfterCandidateIsolation(t *testing.T) {
 	c := createJetStreamClusterExplicit(t, "R3S", 3)
 	defer c.shutdown()
 	c.waitOnLeader()
@@ -488,32 +488,38 @@ func TestNRGCandidateStepsDownAfterAE(t *testing.T) {
 	rg := c.createRaftGroup("TEST", 3, newStateAdder)
 	rg.waitOnLeader()
 
-	// Pick a random follower node. Bump the term up by a considerable
+	// Bump the term up on one of the follower nodes by a considerable
 	// amount and force it into the candidate state. This is what happens
 	// after a period of time in isolation.
-	n := rg.nonLeader().node().(*raft)
-	n.Lock()
-	n.term += 100
-	n.switchState(Candidate)
-	n.Unlock()
+	follower := rg.nonLeader().node().(*raft)
+	follower.Lock()
+	follower.term += 100
+	follower.switchState(Candidate)
+	follower.Unlock()
 
-	// Have the leader push through something on the current term just
-	// for good measure, although the heartbeats probably work too.
+	follower.requestVote()
+	time.Sleep(time.Millisecond * 100)
+
+	// The candidate will shortly send a vote request. When that happens,
+	// the rest of the nodes in the cluster should move up to that term,
+	// even though they will not grant the vote.
+	nterm := follower.term
+	for _, n := range rg {
+		require_Equal(t, n.node().Term(), nterm)
+	}
+
+	// Have the leader send out a proposal, which will force the candidate
+	// back into follower state.
+	rg.waitOnLeader()
 	rg.leader().(*stateAdder).proposeDelta(1)
+	rg.waitOnTotal(t, 1)
 
-	// Wait for the leader to receive the next append entry from the
-	// current leader. What should happen is that the node steps down
-	// and starts following the leader, as nothing in the log of the
-	// follower is newer than the term of the leader.
-	checkFor(t, time.Second, 50*time.Millisecond, func() error {
-		if n.State() == Candidate {
-			return fmt.Errorf("shouldn't still be candidate state")
-		}
-		if nterm, lterm := n.Term(), rg.leader().node().Term(); nterm != lterm {
-			return fmt.Errorf("follower term %d should match leader term %d", nterm, lterm)
-		}
-		return nil
-	})
+	// The candidate should have switched to a follower on a term equal to
+	// or newer than the candidate had.
+	for _, n := range rg {
+		require_NotEqual(t, n.node().State(), Candidate)
+		require_True(t, n.node().Term() >= nterm)
+	}
 }
 
 // Test to make sure this does not cause us to truncate our wal or enter catchup state.
@@ -635,3 +641,42 @@ func TestNRGLeavesObserverAfterPause(t *testing.T) {
 	n.ResumeApply()
 	checkState(false, false)
 }
+
+func TestNRGCandidateDoesntRevertTermAfterOldAE(t *testing.T) {
+	c := createJetStreamClusterExplicit(t, "R3S", 3)
+	defer c.shutdown()
+
+	rg := c.createMemRaftGroup("TEST", 3, newStateAdder)
+	rg.waitOnLeader()
+
+	// Bump the term up a few times.
+	for i := 0; i < 3; i++ {
+		rg.leader().node().StepDown()
+		time.Sleep(time.Millisecond * 50) // Needed because stepdowns not synchronous
+		rg.waitOnLeader()
+	}
+
+	leader := rg.leader().node().(*raft)
+	follower := rg.nonLeader().node().(*raft)
+
+	// Sanity check that we are where we expect to be.
+	require_Equal(t, leader.term, 4)
+	require_Equal(t, follower.term, 4)
+
+	// At this point the active term is 4 and pterm is 4, force the
+	// term up to 9. This won't bump the pterm.
+	rg.lockAll()
+	for _, n := range rg {
+		n.node().(*raft).term += 5
+	}
+	rg.unlockAll()
+
+	// Build an AE that has a term newer than the pterm but older than
+	// the term. Give it to the follower in candidate state.
+	ae := newAppendEntry(leader.id, 6, leader.commit, leader.pterm, leader.pindex, nil)
+	follower.switchToCandidate()
+	follower.processAppendEntry(ae, nil)
+
+	// The candidate must not have reverted back to term 6.
+	require_NotEqual(t, follower.term, 6)
+}