Repository: tomarrell/lbadd
Pull request #129 (2020-05-19 04:33:49)

raft: implements appendEntriesResponse
When appendEntry RPC is sent by leader to other nodes, this function generate response for the leader and handles the entries that should be stored in node logs.


Patch:
diff --git a/internal/raft/append_entries.go b/internal/raft/append_entries.go
index 451cdc2..0768b03 100644
--- a/internal/raft/append_entries.go
+++ b/internal/raft/append_entries.go
@@ -6,6 +6,45 @@ import "github.com/tomarrell/lbadd/internal/raft/message"
 // to the follower node. This function generates the response to be sent to the leader node.
 // This is the response to the contact by the leader to assert it's leadership.
 func AppendEntriesResponse(node *Node, req *message.AppendEntriesRequest) *message.AppendEntriesResponse {
+	leaderTerm := req.GetTerm()
+	nodePersistentState := node.PersistentState
+	nodeTerm := nodePersistentState.CurrentTerm
+	// return false if term is greater than currentTerm
+	// return false if msg Log Index is greater than node commit Index
+	// return false if term of msg at PrevLogIndex doesn't match prev Log Term stored by Leader
+	if nodeTerm > leaderTerm ||
+		req.GetPrevLogIndex() > node.VolatileState.CommitIndex ||
+		nodePersistentState.Log[req.PrevLogIndex].Term != req.GetPrevLogTerm() {
+		return &message.AppendEntriesResponse{
+			Term:    nodeTerm,
+			Success: false,
+		}
+	}
+
+	entries := req.GetEntries()
+	// if heartbeat, skip adding entries
+	if len(entries) > 0 {
+		nodePersistentState.mu.Lock()
+		if req.GetPrevLogIndex() < node.VolatileState.CommitIndex {
+			node.PersistentState.Log = node.PersistentState.Log[:req.GetPrevLogIndex()]
+		}
+		node.PersistentState.Log = append(node.PersistentState.Log, entries...)
+		node.PersistentState.mu.Unlock()
+	}
+
+	if req.GetLeaderCommit() > node.VolatileState.CommitIndex {
+		nodeCommitIndex := req.GetLeaderCommit()
+		if int(req.GetLeaderCommit()) > len(node.PersistentState.Log) {
+			nodeCommitIndex = int32(len(node.PersistentState.Log))
+		}
+		node.VolatileState.CommitIndex = nodeCommitIndex
+		// TODO: Issue #152 apply the log command & update lastApplied
+	}
+
+	return &message.AppendEntriesResponse{
+		Term:    nodeTerm,
+		Success: true,
+	}
 
-	return nil
 }
+


Test patch:
diff --git a/internal/raft/append_entries_test.go b/internal/raft/append_entries_test.go
new file mode 100644
index 0000000..ab34323
--- /dev/null
+++ b/internal/raft/append_entries_test.go
@@ -0,0 +1,95 @@
+package raft
+
+import (
+	"net"
+	"testing"
+
+	"github.com/rs/zerolog"
+	"github.com/stretchr/testify/assert"
+	"github.com/tomarrell/lbadd/internal/network"
+	"github.com/tomarrell/lbadd/internal/raft/cluster"
+	"github.com/tomarrell/lbadd/internal/raft/message"
+)
+
+// TestAppendEntries function checks the correctnes of AppendEntriesResponse
+// function. In this test function, we check how the function will respond to
+// if node Term is less than leader node, node Log Index is less than leader
+// commitIndex and checks if logs are appended correctly to node Log.
+func TestAppendEntries(t *testing.T) {
+	assert := assert.New(t)
+
+	log := zerolog.Nop()
+	cluster := cluster.NewTCPCluster(log)
+
+	conn1, conn2 := net.Pipe()
+	conn3, conn4 := net.Pipe()
+	tcp1int, tcp1ext := network.NewTCPConn(conn1), network.NewTCPConn(conn2)
+	tcp2int, tcp2ext := network.NewTCPConn(conn3), network.NewTCPConn(conn4)
+	defer func() {
+		_ = tcp1int.Close()
+		_ = tcp1ext.Close()
+		_ = tcp2int.Close()
+		_ = tcp2ext.Close()
+	}()
+	cluster.AddConnection(tcp1int)
+	cluster.AddConnection(tcp2int)
+
+	// Created a mock node with default values for PersistentState
+	// and Volatile State.
+	// For Volatile State, CommitIndex given -1 to show no commit is
+	// applied as Log Index start with 0, so -1 show value before any
+	// log committed and same logic goes with LastApplied as -1 show
+	// no logs with given Index is applied to State Machine.
+	node := &Node{
+		State: StateFollower.String(),
+		PersistentState: &PersistentState{
+			CurrentTerm: 0,
+			VotedFor:    nil,
+			SelfID:      cluster.OwnID(),
+			PeerIPs:     cluster.Nodes(),
+		},
+		VolatileState: &VolatileState{
+			CommitIndex: -1,
+			LastApplied: -1,
+		},
+	}
+
+	entries := []*message.LogData{
+		message.NewLogData(2, "execute cmd3"),
+		message.NewLogData(2, "execute cmd4"),
+	}
+	// Creating a mock msg AppendEntriesRequest with default values
+	// Leader commit specifies the Index of Log commited by leader and
+	// entries include msg LogData sent to nodes.
+	msg := &message.AppendEntriesRequest{
+		Term:         1,
+		PrevLogIndex: -1,
+		PrevLogTerm:  1,
+		Entries:      entries,
+		LeaderCommit: 3,
+	}
+
+	node.PersistentState.CurrentTerm = 3
+	res := AppendEntriesResponse(node, msg)
+	assert.False(res.Success, "Node Term is lesser than leader term")
+	msg.Term = 3
+	msg.PrevLogIndex = 3
+	node.VolatileState.CommitIndex = 2
+	res = AppendEntriesResponse(node, msg)
+	assert.False(res.Success, "Node Log Index is lesser than"+
+		"leader commit Index")
+	msg.Term = 2
+	node.PersistentState.CurrentTerm = 2
+	msg.PrevLogIndex = 1
+	msg.PrevLogTerm = 1
+	node.VolatileState.CommitIndex = 1
+	node.PersistentState.Log = []*message.LogData{message.NewLogData(1,
+		"execute cmd1"), message.NewLogData(1, "execute cmd2")}
+	numberOfPersistentLog := len(node.PersistentState.Log)
+	res = AppendEntriesResponse(node, msg)
+	assert.True(res.Success, "Msg isn't appended to the node Logs")
+	assert.Equal(node.PersistentState.CurrentTerm, res.GetTerm(),
+		"Node doesn't have same term as leader")
+	assert.Equal(len(node.PersistentState.Log),
+		numberOfPersistentLog+len(entries), "LogData hasn't appended to the node ")
+}