Refactor processAppendEntry()

Simplify processAppendEntryLocked. The method is no longer used
for the recovery case, i.e. where `sub == nil`. We can simplify
if statements where we check if sub is nil or sub not nil.

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

Author: sciascid

server/raft.go

@@ -3483,8 +3483,8 @@ func (n *raft) processAppendEntryLocked(ae *appendEntry, sub *subscription) *app
 		n.resetElectionTimeout()
 	}
 
-	// Just return if closed or we had previous write error.
-	if n.State() == Closed || n.werr != nil {
+	// Just return if closed or we had previous write error, or invalid sub
+	if n.State() == Closed || n.werr != nil || sub == nil {
 		return nil
 	}
 
@@ -3539,7 +3539,7 @@ func (n *raft) processAppendEntryLocked(ae *appendEntry, sub *subscription) *app
 	catchingUp := n.catchup != nil
 	// Is this a new entry? New entries will be delivered on the append entry
 	// sub, rather than a catch-up sub.
-	isNew := sub != nil && sub == n.aesub
+	isNew := sub == n.aesub
 
 	// Track leader directly
 	if isNew && ae.leader != noLeader {
@@ -3553,7 +3553,7 @@ func (n *raft) processAppendEntryLocked(ae *appendEntry, sub *subscription) *app
 	// If we are/were catching up ignore old catchup subs, but only if catching up from an older server
 	// that doesn't send the leader term when catching up. We can reject old catchups from newer subs
 	// later, just by checking the append entry is on the correct term.
-	if !isNew && sub != nil && ae.lterm == 0 && (!catchingUp || sub != n.catchup.sub) {
+	if !isNew && ae.lterm == 0 && (!catchingUp || sub != n.catchup.sub) {
 		n.debug("AppendEntry ignoring old entry from previous catchup")
 		return nil
 	}
@@ -3569,9 +3569,8 @@ func (n *raft) processAppendEntryLocked(ae *appendEntry, sub *subscription) *app
 			n.debug("Term higher than ours and we are not a follower: %v, stepping down to %q", n.State(), ae.leader)
 			n.stepdownLocked(ae.leader)
 		}
-	} else if lterm < n.term && sub != nil && (isNew || ae.lterm != 0) {
+	} else if lterm < n.term && (isNew || ae.lterm != 0) {
 		// Anything that's below our expected highest term needs to be rejected.
-		// Unless we're replaying (sub=nil), in which case we'll always continue.
 		// For backward-compatibility we shouldn't reject if we're being caught up by an old server.
 		if isNew {
 			n.debug("Rejected AppendEntry from a leader (%s) with term %d which is less than ours", ae.leader, lterm)
@@ -3729,21 +3728,14 @@ func (n *raft) processAppendEntryLocked(ae *appendEntry, sub *subscription) *app
 CONTINUE:
 	// Save to our WAL if we have entries.
 	if ae.shouldStore() {
-		// Only store if an original which will have sub != nil
-		if sub != nil {
-			if err := n.storeToWAL(ae); err != nil {
-				if err != ErrStoreClosed {
-					n.warn("Error storing entry to WAL: %v", err)
-				}
-				return nil
+		if err := n.storeToWAL(ae); err != nil {
+			if err != ErrStoreClosed {
+				n.warn("Error storing entry to WAL: %v", err)
 			}
-			n.cachePendingEntry(ae)
-			n.resetInitializing()
-		} else {
-			// This is a replay on startup so just take the appendEntry version.
-			n.pterm = ae.term
-			n.pindex = ae.pindex + 1
+			return nil
 		}
+		n.cachePendingEntry(ae)
+		n.resetInitializing()
 	}
 
 	// ae should no longer be used after this call as
@@ -3753,7 +3745,7 @@ CONTINUE:
 	// Only ever respond to new entries.
 	// Never respond to catchup messages, because providing quorum based on this is unsafe.
 	// The only way for the leader to receive "success" MUST be through this path.
-	if sub != nil && isNew {
+	if isNew {
 		// Success. Send our response.
 		return newAppendEntryResponse(n.pterm, n.pindex, n.id, true)
 	}