NRG (2.11): Heartbeat can establish quorum & reset acks during stepdown (#6512)

If the leader changed, a heartbeat would not be able to signal there's
quorum on messages. This is because the leader only tracked quorum for
entries it sent out itself AND were stored in its own log. Heartbeats
are sent by the leader but are not stored in its log, which means they
can't be used to track quorum.

To solve this we don't need to track the leader in `n.acks`, which would
not get populated after a leader change. Instead we only track the
replicas that signal success and we always count ourselves anyway. This
ensures that heartbeats, although not stored, can still signal quorum
properly.

Also, acks from previous terms would not be reset during stepdown. Which
meant that successful acks from previous terms could be combined with
acks from newer terms, making the leader believe it has quorum whereas
it didn't properly check all of those acks came from the same term.

Signed-off-by: Maurice van Veen <[email protected]>

Author: derekcollison

server/raft.go

@@ -3020,18 +3020,23 @@ func (n *raft) trackResponse(ar *appendEntryResponse) {
 	// See if we have items to apply.
 	var sendHB bool
 
-	if results := n.acks[ar.index]; results != nil {
-		results[ar.peer] = struct{}{}
-		if nr := len(results); nr >= n.qn {
-			// We have a quorum.
-			for index := n.commit + 1; index <= ar.index; index++ {
-				if err := n.applyCommit(index); err != nil && err != errNodeClosed {
-					n.error("Got an error applying commit for %d: %v", index, err)
-					break
-				}
+	results := n.acks[ar.index]
+	if results == nil {
+		results = make(map[string]struct{})
+		n.acks[ar.index] = results
+	}
+	results[ar.peer] = struct{}{}
+
+	// We don't count ourselves to account for leader changes, so add 1.
+	if nr := len(results); nr+1 >= n.qn {
+		// We have a quorum.
+		for index := n.commit + 1; index <= ar.index; index++ {
+			if err := n.applyCommit(index); err != nil && err != errNodeClosed {
+				n.error("Got an error applying commit for %d: %v", index, err)
+				break
 			}
-			sendHB = n.prop.len() == 0
 		}
+		sendHB = n.prop.len() == 0
 	}
 	n.Unlock()
 
@@ -3765,8 +3770,6 @@ func (n *raft) sendAppendEntry(entries []*Entry) {
 		if err := n.storeToWAL(ae); err != nil {
 			return
 		}
-		// We count ourselves.
-		n.acks[n.pindex] = map[string]struct{}{n.id: {}}
 		n.active = time.Now()
 
 		// Save in memory for faster processing during applyCommit.
@@ -4281,6 +4284,10 @@ func (n *raft) switchToFollowerLocked(leader string) {
 
 	n.aflr = 0
 	n.lxfer = false
+	// Reset acks, we can't assume acks from a previous term are still valid in another term.
+	if len(n.acks) > 0 {
+		n.acks = make(map[uint64]map[string]struct{})
+	}
 	n.updateLeader(leader)
 	n.switchState(Follower)
 }

server/raft_test.go

@@ -2000,6 +2000,128 @@ func TestNRGHealthCheckWaitForPendingCommitsWhenPaused(t *testing.T) {
 	require_True(t, n.Healthy())
 }
 
+func TestNRGHeartbeatCanEstablishQuorumAfterLeaderChange(t *testing.T) {
+	n, cleanup := initSingleMemRaftNode(t)
+	defer cleanup()
+
+	// Create a sample entry, the content doesn't matter, just that it's stored.
+	esm := encodeStreamMsgAllowCompress("foo", "_INBOX.foo", nil, nil, 0, 0, true)
+	entries := []*Entry{newEntry(EntryNormal, esm)}
+
+	nats0 := "S1Nunr6R" // "nats-0"
+
+	// Timeline
+	aeMsg := encode(t, &appendEntry{leader: nats0, term: 1, commit: 0, pterm: 0, pindex: 0, entries: entries})
+	aeHeartbeatResponse := &appendEntryResponse{term: 1, index: 1, peer: nats0, success: true}
+
+	// Process first message.
+	n.processAppendEntry(aeMsg, n.aesub)
+	require_Equal(t, n.pindex, 1)
+	require_Equal(t, n.aflr, 0)
+
+	// Simulate becoming leader, and not knowing if the stored entry has quorum and can be committed.
+	// Switching to leader should send a heartbeat.
+	n.switchToLeader()
+	require_Equal(t, n.aflr, 1)
+	require_Equal(t, n.commit, 0)
+
+	// We simulate receiving the successful heartbeat response here. It should move the commit up.
+	n.processAppendEntryResponse(aeHeartbeatResponse)
+	require_Equal(t, n.commit, 1)
+	require_Equal(t, n.aflr, 1)
+
+	// Once the entry is applied, it should reset the applied floor.
+	n.Applied(1)
+	require_Equal(t, n.aflr, 0)
+}
+
+func TestNRGQuorumAccounting(t *testing.T) {
+	n, cleanup := initSingleMemRaftNode(t)
+	defer cleanup()
+
+	// Create a sample entry, the content doesn't matter, just that it's stored.
+	esm := encodeStreamMsgAllowCompress("foo", "_INBOX.foo", nil, nil, 0, 0, true)
+	entries := []*Entry{newEntry(EntryNormal, esm)}
+
+	nats1 := "yrzKKRBu" // "nats-1"
+	nats2 := "cnrtt3eg" // "nats-2"
+
+	// Timeline
+	aeHeartbeat1Response := &appendEntryResponse{term: 1, index: 1, peer: nats1, success: true}
+	aeHeartbeat2Response := &appendEntryResponse{term: 1, index: 1, peer: nats2, success: true}
+
+	// Adjust cluster size, so we need at least 2 responses from other servers to establish quorum.
+	require_NoError(t, n.AdjustBootClusterSize(5))
+	require_Equal(t, n.csz, 5)
+	require_Equal(t, n.qn, 3)
+
+	// Switch this node to leader, and send an entry.
+	n.switchToLeader()
+	require_Equal(t, n.pindex, 0)
+	n.sendAppendEntry(entries)
+	require_Equal(t, n.pindex, 1)
+
+	// The first response MUST NOT indicate quorum has been reached.
+	n.processAppendEntryResponse(aeHeartbeat1Response)
+	require_Equal(t, n.commit, 0)
+
+	// The second response means we have reached quorum and can move commit up.
+	n.processAppendEntryResponse(aeHeartbeat2Response)
+	require_Equal(t, n.commit, 1)
+}
+
+func TestNRGRevalidateQuorumAfterLeaderChange(t *testing.T) {
+	n, cleanup := initSingleMemRaftNode(t)
+	defer cleanup()
+
+	// Create a sample entry, the content doesn't matter, just that it's stored.
+	esm := encodeStreamMsgAllowCompress("foo", "_INBOX.foo", nil, nil, 0, 0, true)
+	entries := []*Entry{newEntry(EntryNormal, esm)}
+
+	nats1 := "yrzKKRBu" // "nats-1"
+	nats2 := "cnrtt3eg" // "nats-2"
+
+	// Timeline
+	aeHeartbeat1Response := &appendEntryResponse{term: 1, index: 1, peer: nats1, success: true}
+	aeHeartbeat2Response := &appendEntryResponse{term: 1, index: 1, peer: nats2, success: true}
+
+	// Adjust cluster size, so we need at least 2 responses from other servers to establish quorum.
+	require_NoError(t, n.AdjustBootClusterSize(5))
+	require_Equal(t, n.csz, 5)
+	require_Equal(t, n.qn, 3)
+
+	// Switch this node to leader, and send an entry.
+	n.term++
+	n.switchToLeader()
+	require_Equal(t, n.term, 1)
+	require_Equal(t, n.pindex, 0)
+	n.sendAppendEntry(entries)
+	require_Equal(t, n.pindex, 1)
+
+	// We have one server that signals the message was stored. The leader will add 1 to the acks count.
+	n.processAppendEntryResponse(aeHeartbeat1Response)
+	require_Equal(t, n.commit, 0)
+	require_Len(t, len(n.acks), 1)
+
+	// We stepdown now and don't know if we will have quorum on the first entry.
+	n.stepdown(noLeader)
+
+	// Let's assume there are a bunch of leader elections now, data being added to the log, being truncated, etc.
+	// We don't know what happened, maybe we were partitioned, but we can't know for sure if the first entry has quorum.
+
+	// We now become leader again.
+	n.term = 6
+	n.switchToLeader()
+	require_Equal(t, n.term, 6)
+
+	// We now receive a successful response from another server saying they have stored it.
+	// Anything can have happened to the replica that said success before, we can't assume that's still valid.
+	// So our commit must stay the same and we restart counting for quorum.
+	n.processAppendEntryResponse(aeHeartbeat2Response)
+	require_Equal(t, n.commit, 0)
+	require_Len(t, len(n.acks), 1)
+}
+
 // This is a RaftChainOfBlocks test where a block is proposed and then we wait for all replicas to apply it before
 // proposing the next one.
 // The test may fail if: