NRG (2.11): Truncate entries without quorum from different pterm

server/raft.go

@@ -3151,10 +3151,10 @@ func (n *raft) catchupStalled() bool {
 	if n.catchup == nil {
 		return false
 	}
-	if n.catchup.pindex == n.commit {
+	if n.catchup.pindex == n.pindex {
 		return time.Since(n.catchup.active) > 2*time.Second
 	}
-	n.catchup.pindex = n.commit
+	n.catchup.pindex = n.pindex
 	n.catchup.active = time.Now()
 	return false
 }
@@ -3173,7 +3173,7 @@ func (n *raft) createCatchup(ae *appendEntry) string {
 		cterm:  ae.pterm,
 		cindex: ae.pindex,
 		pterm:  n.pterm,
-		pindex: n.commit,
+		pindex: n.pindex,
 		active: time.Now(),
 	}
 	inbox := n.newCatchupInbox()
@@ -3343,7 +3343,7 @@ func (n *raft) processAppendEntry(ae *appendEntry, sub *subscription) {
 			if n.catchupStalled() {
 				n.debug("Catchup may be stalled, will request again")
 				inbox = n.createCatchup(ae)
-				ar = newAppendEntryResponse(n.pterm, n.commit, n.id, false)
+				ar = newAppendEntryResponse(n.pterm, n.pindex, n.id, false)
 			}
 			n.Unlock()
 			if ar != nil {
@@ -3383,32 +3383,43 @@ func (n *raft) processAppendEntry(ae *appendEntry, sub *subscription) {
 		n.updateLeadChange(false)
 	}
 
-	if (isNew && ae.pterm != n.pterm) || ae.pindex != n.pindex {
-		// Check if this is a lower index than what we were expecting.
-		if ae.pindex < n.pindex {
-			n.debug("AppendEntry detected pindex less than ours: %d:%d vs %d:%d", ae.pterm, ae.pindex, n.pterm, n.pindex)
+	if ae.pterm != n.pterm || ae.pindex != n.pindex {
+		// Check if this is a lower or equal index than what we were expecting.
+		if ae.pindex <= n.pindex {
+			n.debug("AppendEntry detected pindex less than/equal to ours: %d:%d vs %d:%d", ae.pterm, ae.pindex, n.pterm, n.pindex)
 			var ar *appendEntryResponse
-
-			// An AppendEntry is stored at seq=ae.pindex+1. This can be checked when eae != nil, eae.pindex==ae.pindex.
-			seq := ae.pindex + 1
 			var success bool
-			if eae, _ := n.loadEntry(seq); eae == nil {
+
+			if n.commit > 0 && ae.pindex <= n.commit {
+				// Check if only our terms do not match here.
+				if ae.pindex == n.pindex {
+					// Make sure pterms match and we take on the leader's.
+					// This prevents constant spinning.
+					n.truncateWAL(ae.pterm, ae.pindex)
+				} else {
+					// If we have already committed this entry, just mark success.
+					success = true
+				}
+			} else if eae, _ := n.loadEntry(ae.pindex); eae == nil {
 				// If terms are equal, and we are not catching up, we have simply already processed this message.
 				// So we will ACK back to the leader. This can happen on server restarts based on timings of snapshots.
 				if ae.pterm == n.pterm && !catchingUp {
 					success = true
+				} else if ae.pindex == n.pindex {
+					// Check if only our terms do not match here.
+					// Make sure pterms match and we take on the leader's.
+					// This prevents constant spinning.
+					n.truncateWAL(ae.pterm, ae.pindex)
 				} else {
 					n.resetWAL()
 				}
-			} else if eae.term != ae.term {
+			} else {
 				// If terms mismatched, delete that entry and all others past it.
 				// Make sure to cancel any catchups in progress.
 				// Truncate will reset our pterm and pindex. Only do so if we have an entry.
 				n.truncateWAL(eae.pterm, eae.pindex)
-			} else {
-				success = true
 			}
-			// Cancel regardless if truncated/unsuccessful.
+			// Cancel regardless if unsuccessful.
 			if !success {
 				n.cancelCatchup()
 			}
@@ -3434,16 +3445,6 @@ func (n *raft) processAppendEntry(ae *appendEntry, sub *subscription) {
 				return
 			}
 
-			// Check if only our terms do not match here.
-			if ae.pindex == n.pindex {
-				// Make sure pterms match and we take on the leader's.
-				// This prevents constant spinning.
-				n.truncateWAL(ae.pterm, ae.pindex)
-				n.cancelCatchup()
-				n.Unlock()
-				return
-			}
-
 			if ps, err := decodePeerState(ae.entries[1].Data); err == nil {
 				n.processPeerState(ps)
 				// Also need to copy from client's buffer.
@@ -3483,19 +3484,16 @@ func (n *raft) processAppendEntry(ae *appendEntry, sub *subscription) {
 			n.apply.push(newCommittedEntry(n.commit, ae.entries[:1]))
 			n.Unlock()
 			return
-
-		} else {
-			n.debug("AppendEntry did not match %d %d with %d %d (commit %d)", ae.pterm, ae.pindex, n.pterm, n.pindex, n.commit)
-			if ae.pindex > n.commit {
-				// Setup our state for catching up.
-				inbox := n.createCatchup(ae)
-				ar := newAppendEntryResponse(n.pterm, n.commit, n.id, false)
-				n.Unlock()
-				n.sendRPC(ae.reply, inbox, ar.encode(arbuf))
-				arPool.Put(ar)
-				return
-			}
 		}
+
+		// Setup our state for catching up.
+		n.debug("AppendEntry did not match %d %d with %d %d", ae.pterm, ae.pindex, n.pterm, n.pindex)
+		inbox := n.createCatchup(ae)
+		ar := newAppendEntryResponse(n.pterm, n.pindex, n.id, false)
+		n.Unlock()
+		n.sendRPC(ae.reply, inbox, ar.encode(arbuf))
+		arPool.Put(ar)
+		return
 	}
 
 	// Save to our WAL if we have entries.

server/raft_helpers_test.go

@@ -317,3 +317,34 @@ func (rg smGroup) waitOnTotal(t *testing.T, expected int64) {
 func newStateAdder(s *Server, cfg *RaftConfig, n RaftNode) stateMachine {
 	return &stateAdder{s: s, n: n, cfg: cfg, lch: make(chan bool, 1)}
 }
+
+func initSingleMemRaftNode(t *testing.T) (*raft, func()) {
+	t.Helper()
+	c := createJetStreamClusterExplicit(t, "R3S", 3)
+	s := c.servers[0] // RunBasicJetStreamServer not available
+
+	ms, err := newMemStore(&StreamConfig{Name: "TEST", Storage: MemoryStorage})
+	require_NoError(t, err)
+	cfg := &RaftConfig{Name: "TEST", Store: t.TempDir(), Log: ms}
+
+	err = s.bootstrapRaftNode(cfg, nil, false)
+	require_NoError(t, err)
+	n, err := s.initRaftNode(globalAccountName, cfg, pprofLabels{})
+	require_NoError(t, err)
+
+	cleanup := func() {
+		c.shutdown()
+	}
+	return n, cleanup
+}
+
+// Encode an AppendEntry.
+// An AppendEntry is encoded into a buffer and that's stored into the WAL.
+// This is a helper function to generate that buffer.
+func encode(t *testing.T, ae *appendEntry) *appendEntry {
+	t.Helper()
+	buf, err := ae.encode(nil)
+	require_NoError(t, err)
+	ae.buf = buf
+	return ae
+}