Leader election and Log replication in Raft

Raft is a consensus algorithm for managing a replicated log. This write-up will go over Leader election and Log Replication. The insights are from finishing assignments 3A, 3B, 3C from the MIT 6.824 Distributed Systems course. This post assumes that you have read the Raft paper, ideally multiple times.

Leader election

Raft enforces strong leadership. It elects one leader which is responsible for sending log entries to all other servers (peers). Log entries always flow from leader to followers.

Every raft server has an electionTimeout (stored persistently on disk) that is randomly set from within a range for e.g., between 150-300ms. Elections are managed using RequestVote RPCs. Below is the pseudo-code for how elections occur and how a leader is elected.

Every peer has a long running task (conceptually a cron job)

// The candidate sends a RequestVote RPC to the peer
// this method runs on the peer
// The result of the execution (whether vote was granted etc)
// is sent back via the reply
func RequestVoteRpc(args RequestVoteArgs, reply RequestVoteReply){
   if args.Term < currentTerm {
    reply.Success = false
    return
   }

   reply.Success = false
   if votedFor == null or votedFor == args.CandidateId && IsCandidateLogUpToDate(lastLogIdx, lastLogTerm int) {
    reply.Success = true
   }
}


func startElection(){
    if (time elapsed since receiving AppendEntries RPC or since voting for a candidate > electionTimeout){
        currentState := Candidate
        currentTerm += 1
        voteFor := myself
        votes := 1
        electionTimeout := 150 + (randInit()%150) // reset election timer
        // the RequestVote RPCs are sent in parallel to all peers by the candidate
        for p := peers {
            if p != myself {
                requestVoteReply := go RequestVoteRpc()
                if (requestVoteReply != successful) {
                    sleep(smallDuration)
                    continue // retry the rpc
                } else {
                    votes += 1
                }
            }
        }

        if votes > majority of peers {
            currentState := Leader
            for p := peers {
                if p != myself {
                    go SendAppendEntriesHeartbeat(p)
                }
            }
        }
    }
}

func electionTicker(){
    for {
        sleepDuration := 150 + (randInt()%150)
        Sleep(sleepDuration)
        startElection()
    }
}

At its core for a peer to be elected as a leader, first a peer transitions to candidate, then it sends RequestVote RPCs in parallel to all other peers. If it gets a majority of votes, it becomes a Leader. Not every candidate can become a leader, only candidates that have an up to date log can become leaders, if a candidate doesn’t have an up to date log it won’t receive a majority of votes. This restriction is implemented via the IsCandidateUpToDate method.

Upon election as a leader, the leader sends an AppendEntries RPC to all other peers to establish and maintain it leadership. This is to prevent other peers from attempting to become candidates and also to transfer logs from leader to followers. We will discuss this in the Log replication section.

Log replication

In Raft, all clients (that send logs) talk to the leader only. If a client connects to a non-leader its redirected to the leader. When a client sends a new log entry to be appended, the leader would add it to its own log and then the AppendEntries heartbeat would take care of replicating this to all the peers.

After a leader is elected, it sends an AppendEntries RPC to all other peers. This has two functions, to establish and maintain leadership, for which we send the RPC at a regular cadence and second, if there are new log entries added to the leader’s log this would be replicated by the RPC as well. Although the Raft paper maintains that for an AppendEntries heartbeat, the log entries sent should be empty. This is an implementation detail. In our case, when we send AppendEntries heartbeat and in between the heartbeats if there are logs added to the leader, we would send the newly added logs in the next heartbeat.

// this runs on the peer, args is sent by the leader and reply
// is set by the peer for the leader to consume
func AppendEntriesRpc(peer, appendEntriesArgs, appendEntriesReply){
    if appendEntriesArgs.Term < currentTerm {
        // rpc sent by outdated leader
        // decline the rpc (reply false)
        // Leader will convert to follower on receipt of this RPC response
    }

    if appendEntriesArgs.PrevLogIdx >= len(log) {
        // Leader's log is longer than the follower
        // Decline the RPC (reply false)
        // Leader will update nextIdx after receipt of this rpc response
    }

    if log[appendEntriesArgs.PrevLogidx] != appendEntriesArgs.PrevLogTerm {
        // There is a term mismatch at lastLogIdx b/w leader and follower
        // Decline the RPC (reply false)
        // Leader will update nextIdx after receipt of this rpc response.
    }

    // Terms match at lastLogIdx for leader and peer.
    i := appendEntriesArgs.PrevLogIdx + 1
    j := 0
    for i < len(log) && j < len(appendEntriesArgs.Entries) {
        // find out where leader and follower log disagree (if at all)
        // let that index be k
        // discard follower log after that index
    }

    // Append Log sent by leader starting from index k
    // since its not present on the follower.
    log = append(log, appendEntriesArgs.Entries[k:])

    // AppendEntriesRPC was successful
    // After leader receives this rpc response and updates
    // nextIdx and matchIdx for this peer, leader and follower logs match
    appendEntriesReply.Success = true

    // Update this peer's commit Index based on
    // leader's commitIndex and this peer's updated log
    if appendEntriesArgs.LeaderCommit > commitIndex {
        commitIndex = min(appendEntriesArgs.LeaderCommmit, len(log))
    }
}

// The leader sends an Append Entries RPC to every peer
// The leader runs this and handles the response from the peer
// This doesn't handle the scenarios where RPC responses
// might be dropped or delivered out of order
func SendAppendEntriesRPC(peer int){
    // After an election rf.nextIdx[peer] = len(log)
    lastLogIdx := rf.nextIdx[peer] - 1 // we are optimistically assuming that peer has
    // log up to lastLogIdx.
    lastLogTerm := rf.log[lastLogIdx].Term
    logEntries := rf.log[lastLogIdx+1:] // if peer has log up to lastLogIdx, we need to send
    // log starting from lastLogIdx+1 to be rpelicated on the peer.

    appendEntriesArgs := {
        Term : currentTerm,
        PrevLogIdx: lastLogIdx,
        PrevLogTerm: lastLogTerm,
        Entries: logEntries
    }

    appendEntriesResp := go AppendEntriesRpc(peer, appendEntriesArgs, appendEntriesReply)
    if appendEntriesResp.Status != Failed {
        //peer was down, resp was dropped, req timed out etc
        retry
    } else {
        if appendEntriesReply.Success {
            // log up to lastLogIdx is present on the peer and it added logEntries to its log
            // update nextIndex and matchIndex
            nextIdx[peer] = lastLogIdx + len(logEntries)
            matchIdx[peer] = lastLogIdx + len(logEntries) - 1
        } else {
            nextIdx[peer] = nextIdx[peer] - 1 // we know that log doesn't match at lastLogIdx
            // so decrement it by 1 and retry
            retry
        }
    }
}

func SendAppendEntriesPeriodically(peer int){
    for {
        // Send Append Entries RPC every 10ms to maintain leadership
        // and replicate logs
        go SendAppendEntriesRPC(peer)
        time.Sleep(10ms)
    }
}

func SendAppendEntriesHeartbeat(){
    for p := range peers {
        if p != leader {
            go SendAppendEntriesPeriodically(p)
        }
    }
}

The Raft protocol leaves a lot of room for interpretation from the sender’s side. The protocol doesn’t specify how to handle out of order RPC replies. Here is one bug that stumped me. In Figure 2 of the paper, its mentioned that matchIndex[] increases monotonically. This means that whenever we update matchIndex for a peer on a successful AppendEntries RPC response, we should ensure matchIndex can only increase.

We use matchIndex to update the commitIndex on the leader, which in the end decides whether log entries up to a certain index are committed and eventually applied to the state machine. When an operation is applied to the state machine, we return a successful response to the client and that means that the log entries up to an index submitted by the client are durable and won’t be overwritten in the future.

If there exists an N such that N > commitIndex, a majority
of matchIndex[i] ≥ N, and log[N].term == currentTerm:
set commitIndex = N

If we were to decrement matchIndex, we would potentially be decrementing commitIndex and uncommitting log entries which were already committed thereby violating Leader completeness and State machine safety properties. When we set commitIndex, it will eventually be used to apply the operation to the state machine via

If commitIndex > lastApplied: increment lastApplied, apply
log[lastApplied] to state machine (§5.3)

As to how the bug can manifest, assume leader sends an Append Entries RPC A and RPC B, both were successful. RPC-A was sent before RPC-B but the leader received them out of order i.e., it processed B before A.

AppendEntries RPC-B
PrevLogIdx: 10
PrevLogTerm: 8
Entries: [11, 12, 13] // indices of the entries

AppendEntriesReply
Success = true
nextIndex[peer] = 13
matchIndex[peer] = 12

When we process an older Append Entries response, we must ensure that it doesn’t decrement matchIndex from a newer Append Entries RPC response. If we process RPC-A’s response in this instance, we would be decrementing nextIndex and matchIndex.

AppendEntries RPC-A
PrevLogIdx: 5
PrevLogTerm: 8
Entries: [6, 7] // indices of the entries

AppendEntriesReply // received after B
Success = true

nextIndex[peer] = 8 // WRONG!
matchIndex[peer] = 7 // WRONG!
// Leader and peer already agree till index 12 (matchIndex[peer] = 12 from RPC-B)
// we are removing entries from the peer which were successfully replicated
// and possibly committed and applied to the state machine.