How Data Is Replicated In Distributed Systems | Raft From Scratch - Part 2
In my previous post we have already discussed how to implement the leader election module of the Raft consensus algorithm. In this post we are going to extend that and implement a complete replication system. If you have not read part one of this raft implementation series, then please read it first.
Understanding Raft By Building It From Scratch - Part 1
Important: One thing I would like to mention is that this is not a production-ready implementation of Raft. This is just a project that I built in order to learn.
Source code for this project can be found here. Contributions and improvements are also welcome.
What Does A Leader Do?
A leader does three main things:
1Accept client requests.
2Send heartbeats to followers.
3Replicate data to followers.
As per our last post discussion, we know that a client sent a heartbeat to its follower. At a regular interval of time using AppendEntriesRPC, and if clients do not receive a heartbeat after some time out, a new election started, and a new leader was elected.
So we have a very simple path whenever the client send any request to our leader; we will first store that request data locally in our leader and then replicate that to the followers via the same heartbeat mechanism. So if a leader has new data to replicate, it will send the data along with the heartbeat, and if it does not have any data, it will send an empty heartbeat.
This client request can be anything; for example, if you are building a distributed key-value store, a client can send the following data to the leader:
1SET name "Sushant"
Our raft will store this exact command somewhere and replicate this to followers as well. Once a sufficient number of followers acknowledge the replication Our key value store can process it.
Raft Log
A raft log is an ordered, appendonly sequence of record that is maintained by every node in the distributed system. It contains two things:
- Term: The term number of leader when it received the command.
- Command: Actual Data
Below image can be very helpful to visualise how it looks like.
Now a very basic mental model is that whenever a command is sent to leader by the client, it will first end it in the local raft log and then replicate it to followers. By doing this, all the nodes in our system will end with the same data. But this is a happy case. What happened when a node goes down and then comes back after some time. Our leader would have gone ahead in time, and it would have had more data than the node that was down. So how are we supposed to send old data to our node that recently came online after cash?
The above image will give you a better idea of what I’m talking about. Our leader log has four entries; there are two followers to which our leader is replicating the data. Follower one is completely in sync because it is online all the time, but due to some reason, follower two went down after the entry “B” was replicated to it. Our follower two is behind our leader, so we need to catch up. This is where I will introduce several important components of the raft.
Our problem can be solved with simple variables; let’s see how.
Commit Index
It is an integer that tells us the highest number of log entries that are safely copied / replicated to the majority of the machines.
Last Applied Index
It is the highest entry that is actually executed by our application. It can be very confusing, so I will make it simple. Let’s say we are building a key value store. Commands from the client will first go to the raft, and the raft will replicate it to the majority of machines and then hand it over to our key value store. Then our value store will actually do its application logic and do whatever it needs to do with that command. Such as storing the key, or deleting a key.
This variable but tells us at what entry of index our commands are handed to our application logic for its work.
Next Index
It is a map-based data structure that stores the mapping of the next log entry that our leader will send to the follower. It is different for every follower because you can see that in our diagram there is no new entry that the leader needs to send to full over one, but our leader has to send entry numbers three and four to our follower, which was lagging behind. So this map will help us store the next entry for every follower.
Match Index
This is the mapping of latest index that has been successfully replicated to each follower. It might be confusing but let’s just proceed and you will understand why this exists. Let’s take the same example from our previous image.
Our leader has following log:
1Index 0 1 2 3 4
2 - A B C D
Follower one is completely in sync:
1Index 0 1 2 3 4
2 - A B C D
But follower two was down for some time:
1Index 0 1 2
2 - A B
So our leader can maintain the following state:
1nextIndex = {
2 follower1: 5,
3 follower2: 3,
4}
5
6matchIndex = {
7 follower1: 4,
8 follower2: 2,
9}
For follower one, matchIndex is 4 because the latest entry we know has been successfully replicated to it is entry 4. For follower two, it is 2 because the latest entry that we know has been replicated is entry 2 and nextIndex is simply one position ahead of that.
So for follower two:
1matchIndex = 2
2nextIndex = 3
This means: “I know that follower two has everything up to index 2, so the next entry I should try sending is index 3.” This is exactly what we need when a follower goes down and comes back later.
Updated Raft Struct
1type Raft struct {
2 mu sync.Mutex
3 state State
4 lastEventTime time.Time
5 votedFor string
6 term int64
7 server *Server
8
9 log []LogEntry
10 // commitIndex is the highest log entry safely copied to a majority of machines.
11 commitIndex int
12 // lastApplied is the highest entry actually executed by the local state machine.
13 lastApplied int
14
15 // Leader Only
16 // nextIndex is the mapping of next log index that leader will send to client.
17 nextIndex map[string]int
18 // matchIndex is the mapping of most latest index that has been replicated to client.
19 matchIndex map[string]int
20
21 // Channel on which commands will be sent to application logic.
22 applyChan chan ApplyMessage
23}
Initializing nextIndex
When a new leader is elected, it doesn’t immediately know exactly how much data every follower has. So we initially assume that every follower has the same log as the leader.
For example, if the leader’s log is:
1Index 0 1 2 3 4
2 - A B C D
then the leader initializes:
1nextIndex[follower] = 5
Why 5? Because nextIndex represents the next log entry that we want to send. Our last log entry is at index 4, so the next possible entry is index 5.
So if our log contains:
10 1 2 3 4
2- A B C D
then len(r.log) is 5.
Therefore:
1nextIndex = 5
Our becameLeader() function is handing it.
1func (r *Raft) becameLeader() {
2 // Some Code
3 for peerID := range r.server.peerRPCs {
4 r.nextIndex[peerID] = len(r.log)
5 // This will be set later when leader sends heartbeat.
6 r.matchIndex[peerID] = 0
7 }
8 // Some Other Code
9}
Replicating Log Entries
Now let’s see how the actual replication happens. We already have the AppendEntries RPC from the leader election implementation. In the previous post we were mostly using it for sending heartbeats. But there is something interesting about this RPC. The same RPC is also responsible for replicating log entries.
So a heartbeat is basically an AppendEntries RPC where:
1entries = []
And when the leader actually has something to replicate:
1entries = [new log entries]
Let’s take our previous example again.
The leader has:
1Leader:
2
3Index 0 1 2 3 4
4 - A B C D
Follower two has been offline:
1Follower 2:
2
3Index 0 1 2
4 - A B
The leader initially thinks the follower might already have everything.
Therefore:
1nextIndex[follower2] = 5
So the leader tries to send an AppendEntries request using:
1prevLogIndex = nextIndex - 1 = 4
The important part here is that the leader is not just blindly sending entry D.
It is also telling the follower: “Before accepting these entries, make sure your log matches mine at this previous position.” So the request contains something like:
1prevLogIndex = 4
2prevLogTerm = 3
3entries = []
The follower now checks whether it even has index 4. But our follower only has entries up to index 2.
1Follower:
2
3Index 0 1 2
4 - A B
So index 4 doesn’t exist. The follower simply returns:
1success = false
Leader Goes Backwards
Now the interesting part happens. The leader receives:
1success = false
This tells the leader that its assumption about the follower’s log was wrong. So it decrements:
1nextIndex[follower]--
Now:
1nextIndex = 4
The leader tries again. This time:
1prevLogIndex = 3
But follower two still doesn’t have index 3. So again:
1success = false
The leader decrements again:
1nextIndex = 3
Now:
1prevLogIndex = 2
And this time the follower does have index 2.
1Leader:
2
3Index 0 1 2 3 4
4 - A B C D
5 ↑
6 prevLogIndex
7
8
9Follower:
10
11Index 0 1 2
12 - A B
13 ↑
14 matching entry
The follower can now check:
1follower.log[2].Term == prevLogTerm
If the term matches, we have found a point where both logs are consistent.
The leader can now send:
1entries = [C, D]
And follower two can append them. After successful replication:
1Follower 2:
2
3Index 0 1 2 3 4
4 - A B C D
Here is the code for that.
1func (r *Raft) sendHeartBeats() {
2 r.mu.Lock()
3 if r.state != Leader {
4 r.mu.Unlock()
5 return
6 }
7 savedTerm := r.term
8 leaderID := r.server.id
9 r.mu.Unlock()
10
11 for peerID, peerRPC := range r.server.peerRPCs {
12 r.replicateToPeer(peerID, peerRPC, savedTerm, leaderID)
13 }
14}
15
16func (r *Raft) replicateToPeer(peerID string, peerRPC pb.RaftServiceClient, savedTerm int64, leaderID string) {
17 r.mu.Lock()
18
19 next := r.nextIndex[peerID]
20 prevLogIndex := next - 1
21 prevLogTerm := r.log[prevLogIndex].Term
22 entries := make([]LogEntry, len(r.log[next:]))
23 copy(entries, r.log[next:])
24 pbEntries := []*pb.LogEntry{}
25 leaderCommit := r.commitIndex
26
27 r.mu.Unlock()
28
29 for _, entry := range entries {
30 pbEntries = append(pbEntries, &pb.LogEntry{
31 Term: entry.Term,
32 Command: entry.Command,
33 })
34 }
35
36 req := &pb.AppendEntriesRequest{
37 Term: savedTerm,
38 LeaderId: leaderID,
39 PrevLogIndex: int64(prevLogIndex),
40 PrevLogTerm: prevLogTerm,
41 Entries: pbEntries,
42 LeaderCommit: int64(leaderCommit),
43 }
44
45 // As of now we will hit clients RPCs synchronously to keep things simple.
46 resp, err := peerRPC.AppendEntries(context.Background(), req)
47 if err == nil {
48 r.mu.Lock()
49
50 if r.state != Leader {
51 r.mu.Unlock()
52 return
53 }
54
55 if resp.Term > savedTerm {
56 r.becameFollower(resp.Term)
57 r.mu.Unlock()
58 return
59 }
60
61 if !resp.Success {
62 if r.nextIndex[peerID] > 1 {
63 r.nextIndex[peerID]--
64 }
65 } else {
66 lastReplicatedIndex := int(req.PrevLogIndex) + len(pbEntries)
67 r.nextIndex[peerID] = lastReplicatedIndex + 1
68 r.matchIndex[peerID] = lastReplicatedIndex
69
70 r.advanceLeaderCommit()
71 }
72
73 r.mu.Unlock()
74 }
75}
Now both logs are identical. This is one of the things I really like about Raft. The leader doesn’t need to send some complicated “here is your entire log, replace everything” message. It keeps trying from different positions until it finds a point where both logs agree, and then sends the missing entries from there.
How Does The Follower Validate Entries?
So far we have seen how the leader finds the correct position. But how does the follower actually decide whether an AppendEntries request is valid? Our follower performs a few checks.
First, it checks the term.
1if req.Term < r.term {
2 return resp, nil
3}
If the leader’s term is older than the follower’s current term, the follower rejects the request. Then we check whether the follower even has the PrevLogIndex that the leader is talking about:
1if req.PrevLogIndex > lastLogIndex {
2 return resp, nil
3}
If the follower doesn’t have that entry, it returns:
1success = false
The leader will then decrement nextIndex and try again. But having the index is not enough. The term of that entry must also match.
1if req.PrevLogTerm != r.log[req.PrevLogIndex].Term {
2 return resp, nil
3}
This is important because two logs can have the same index but contain entries from different terms.
For example:
1Leader:
2
3Index 0 1 2 3
4 - A B C
5Term 0 1 2 3
6
7
8Follower:
9
10Index 0 1 2 3
11 - A B X
12Term 0 1 2 2
Both logs have an entry at index 3. But their terms are different. Therefore they are not consistent. The follower rejects the request. The leader will move backwards until it finds an index where the logs agree.
What If The Follower Has Conflicting Entries?
There is another interesting case. A follower might have an entry at the same index, but that entry might belong to a different term.
For example:
1Leader:
2
3Index 0 1 2 3 4
4 - A B C D
5Term 0 1 1 3 3
6
7
8Follower:
9
10Index 0 1 2 3 4
11 - A B X Y
12Term 0 1 1 2 2
Here the logs match until index 2. At index 3, they have different terms. The follower therefore needs to discard its conflicting entries. This is what we are doing here:
1if index <= int64(len(r.log)-1) {
2 if r.log[index].Term != entry.Term {
3 r.log = r.log[:index]
4 ...
5 }
6}
In simple words:
If an existing entry conflicts with the leader’s entry, remove the conflicting entry and everything after it, then append the leader’s entries.
So after replication, the follower becomes:
1Follower:
2
3Index 0 1 2 3 4
4 - A B C D
5Term 0 1 1 3 3
Now its log is consistent with the leader. Here is compelte code for follower logic.
1func (r *Raft) HandleAppendEntries(req *pb.AppendEntriesRequest) (*pb.AppendEntriesResponse, error) {
2 r.mu.Lock()
3 defer r.mu.Unlock()
4
5 resp := &pb.AppendEntriesResponse{
6 Success: false,
7 Term: r.term,
8 }
9
10 if req.Term < r.term {
11 return resp, nil
12 }
13
14 r.lastEventTime = time.Now()
15
16 if req.Term > r.term {
17 r.becameFollower(req.Term)
18 }
19
20 if r.state != Follower {
21 r.becameFollower(req.Term)
22 }
23
24 lastLogIndex := int64(len(r.log) - 1)
25
26 // Leader is expecting follower to have PrevLogIndex but it does not even have that. So it will reply false. Leader will then
27 // decrement index by 1 and resend the entry.
28 // This will keep going until leader sends entry that was actually present in follower log.
29 if req.PrevLogIndex > lastLogIndex {
30 return resp, nil
31 }
32
33 // Term of pervious log entry should also match.
34 if req.PrevLogTerm != r.log[req.PrevLogIndex].Term {
35 return resp, nil
36 }
37
38 // Appending new entries
39 for i, entry := range req.Entries {
40 index := req.PrevLogIndex + 1 + int64(i)
41
42 // There are existing entries that are conflicting with leader new entries. We need to discard them.
43 if index <= int64(len(r.log)-1) {
44 if r.log[index].Term != entry.Term {
45 r.log = r.log[:index]
46
47 for j := i; j < len(req.Entries); j++ {
48 r.log = append(r.log, LogEntry{
49 Term: req.Entries[j].Term,
50 Command: req.Entries[j].Command,
51 })
52 }
53
54 break
55 }
56
57 continue
58 }
59
60 r.log = append(r.log, LogEntry{
61 Term: entry.Term,
62 Command: entry.Command,
63 })
64 }
65
66 if req.LeaderCommit > int64(r.commitIndex) {
67 r.commitIndex = min(int(req.LeaderCommit), len(r.log)-1)
68 }
69
70 resp.Term = r.term
71 resp.Success = true
72
73 return resp, nil
74}
Commit Index
So far we have successfully copied data between machines. But there is still one very important question.
When is it safe to actually execute the command?
Just because the leader has an entry doesn’t mean that the command should immediately be executed by our application. Imagine a cluster of three nodes:
1Leader → A
2Follower 1 → A
3Follower 2 → X
If the leader crashes at this point, the entry A exists on two machines. Therefore it has been replicated to a majority.
This is where commitIndex comes into the picture. commitIndex tells us the highest log index that the leader knows has been replicated to a majority of the cluster.
For example:
1Leader:
2
3Index 0 1 2 3 4
4 - A B C D
5
6commitIndex = 3
This means entries up to index 3 are considered committed. In our implementation, the leader keeps track of how far every follower has replicated using matchIndex.
For example:
1matchIndex = {
2 follower1: 4,
3 follower2: 3,
4}
Our leader itself already has index 4.
Therefore:
1Index 4 → Leader + Follower 1
2Index 3 → Leader + Follower 1 + Follower 2
Entry 3 is safely replicated to a majority of the cluster.
So the leader can move:
1commitIndex = 3
This is handled by our advanceLeaderCommit() function.
1func (r *Raft) advanceLeaderCommit() {
2 clusterSize := len(r.server.peerRPCs) + 1
3
4 for N := len(r.log) - 1; N > r.commitIndex; N-- {
5 replicatedCount := 1
6
7 for peerID := range r.server.peerRPCs {
8 if r.matchIndex[peerID] >= N {
9 replicatedCount++
10 }
11 }
12
13 if replicatedCount > clusterSize/2 && r.log[N].Term == r.term {
14 r.commitIndex = N
15 break
16 }
17 }
18}
One important thing here is that we also check:
1r.log[N].Term == r.term
We only advance the commit index this way for an entry from the current leader’s term. This is an important part of the Raft algorithm and prevents the leader from incorrectly considering certain entries from previous terms committed just because they happen to exist on a majority.
Last Applied
Now we have another variable:
1lastApplied
Remember, commitIndex and lastApplied are not the same thing.
commitIndex tells us:
“This entry is safely replicated.”
lastApplied tells us:
“This entry has actually been handed over to the application.”
For example:
1commitIndex = 4
2lastApplied = 2
This means entries 3 and 4 are already safely committed, but our application has not processed them yet.
Our applyLoop() handles this:
1if r.lastApplied < r.commitIndex {
2 r.lastApplied++
3
4 index := r.lastApplied
5 command := r.log[index].Command
6
7 r.applyChan <- ApplyMessage{
8 Index: int64(index),
9 Command: command,
10 }
11}
It keeps applying entries one by one until:
1lastApplied == commitIndex
For example:
1commitIndex = 4
2lastApplied = 2
The application receives:
1Entry 3
2Entry 4
After that:
1commitIndex = 4
2lastApplied = 4
Now the application has caught up.
Putting Everything Together
At this point we have quite a few variables:
1commitIndex
2lastApplied
3nextIndex
4matchIndex
But they all have a very simple purpose.
1nextIndex
2 ↓
3"What should I send to this follower next?"
4
5matchIndex
6 ↓
7"What is the latest entry I know this follower has?"
8
9commitIndex
10 ↓
11"What is safely replicated to a majority?"
12
13lastApplied
14 ↓
15"What has actually been processed by the application?"
So the overall flow looks something like this:
1Client
2 |
3 | command
4 ↓
5Leader
6 |
7 | append to local log
8 ↓
9Leader Log
10 |
11 | AppendEntries
12 ↓
13Followers
14 |
15 | successful replication
16 ↓
17matchIndex updated
18 |
19 | majority replicated
20 ↓
21commitIndex updated
22 |
23 | apply committed entries
24 ↓
25Application / State Machine
And this gives us the complete journey of a command. A client sends a command to the leader. The leader first puts that command into its own log. It then uses AppendEntries to replicate that log entry to followers. If a follower is behind, the leader uses nextIndex to find where it should start sending entries. The follower validates the previous log index and term before accepting the new entries. Once the entry has been replicated to a majority, the leader advances commitIndex. Finally, the committed command is sent to the application through applyChan, and lastApplied is updated.
So we have gone from:
1Client → Leader → Log → Followers
to:
1Client
2 ↓
3Leader Log
4 ↓
5Follower Logs
6 ↓
7Majority
8 ↓
9Commit
10 ↓
11Application
And this is basically the core of Raft log replication. There are still several things that can make this implementation much more robust, such as persisting Raft state, handling crashes correctly, optimizing the way conflicting logs are repaired, and dealing with concurrent replication.
But for learning how Raft works internally, I think this gives us a pretty good foundation.
Thanks for reading this long long post.