Refactor processAppendEntry()

Add recoverAppendEntry() to handle append entries during recovery.
This is to separate the normal path from recovery path.

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

Author: sciascid

server/raft.go

@@ -514,7 +514,7 @@ func (s *Server) initRaftNode(accName string, cfg *RaftConfig, labels pprofLabel
 				truncateAndErr(index - 1)
 				break
 			}
-			n.processAppendEntry(ae, nil)
+			n.recoverAppendEntry(ae)
 			// Check how much we have queued up so far to determine if we should pause.
 			for _, e := range ae.entries {
 				qsz += len(e.Data)
@@ -3444,9 +3444,19 @@ func (n *raft) updateLeader(newLeader string) {
 	}
 }
 
-// processAppendEntry will process an appendEntry. This is called either
-// during recovery or from processAppendEntries when there are new entries
-// to be committed.
+// recoverAppendEntry will process an appendEntry for recovery
+func (n *raft) recoverAppendEntry(ae *appendEntry) {
+	n.Lock()
+	defer n.Unlock()
+
+	n.pterm = ae.term
+	n.pindex = ae.pindex + 1
+
+	n.processEntriesAndCommit(ae, false)
+}
+
+// processAppendEntry will process an appendEntry. This is called from
+// processAppendEntries when there are new entries to be committed.
 func (n *raft) processAppendEntry(ae *appendEntry, sub *subscription) {
 	// Make a copy of the reply subject, as ae may return
 	// to its pool as part of processAppendEntryLocked

server/raft_test.go

@@ -807,7 +807,7 @@ func TestNRGCandidateDoesntRevertTermAfterOldAE(t *testing.T) {
 	// 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)
+	follower.processAppendEntry(ae, follower.aesub)
 
 	// The candidate must not have reverted back to term 6.
 	require_NotEqual(t, follower.term, 6)