From Elections to Replication
Part 1 of this series built the election machinery for a Raft cluster. This part extends that implementation so the ConsensusModule can accept commands from a service, replicate those commands across peers, and notify the service when commands are safely committed. The code structure stays the same as before, with new structs and functions added alongside modifications to the existing election logic. All code for this part lives in the part2 directory of the repository.
In Raft's terminology, a client is a service (a key-value store, for instance) that sends commands — arbitrary values of Go's any type — into the consensus module. The service also listens for commit notifications from Raft, which tell it which commands are now safe to apply to its own state machine. There is no requirement that a command be submitted to every peer; the service typically talks only to the leader. Once a command is submitted, it flows through three stages: submission to the leader, replication to followers, and eventual commitment once a majority of peers acknowledge having the command in their logs.
The commit notification path is implemented with a channel. When a ConsensusModule is constructed, it receives commitChan chan<- CommitEntry, where a CommitEntry carries the committed command and its index in the log. A callback would work equally well here; a channel keeps the interface simple and idiomatic Go.
// CommitEntry is the data reported by Raft to the commit channel. Each commit
// entry notifies the client that consensus was reached on a command and it can
// be applied to the client's state machine.
type CommitEntry struct {
// Command is the client command being committed.
Command any
// Index is the log index at which the client command is committed.
Index int
// Term is the Raft term at which the client command is committed.
Term int
}
Log Structure and Command Submission
Each peer maintains a linear log of commands, where every entry records both a term and a command. The terms come from election rounds; the commands are the data that the service wants replicated. Because entries are indexed linearly, the entire state of the service can be reconstructed by replaying the log from the start. In this implementation, a log entry is a small struct, and each peer's log is simply a slice of these entries.
type LogEntry struct {
Command any
Term int
}
Submitting a command is straightforward. The Submit method appends to the leader's log and returns true; on a follower it is ignored and returns false. A true return value is not a guarantee of commitment, though. If the leader is partitioned and a new leader is elected, the old leader may still be accepting submissions from clients that cannot yet see the new leader. A client should wait for its command to appear on the commit channel for a reasonable timeout; if it never appears, the client should retry with another likely leader.
func (cm *ConsensusModule) Submit(command any) bool {
cm.mu.Lock()
defer cm.mu.Unlock()
cm.dlog("Submit received by %v: %v", cm.state, command)
if cm.state == Leader {
cm.log = append(cm.log, LogEntry{Command: command, Term: cm.currentTerm})
cm.dlog("... log=%v", cm.log)
return true
}
return false
}
Two internal mechanisms handle the gap between appending a command to the log and notifying the service of a commit. The first is newCommitReadyChan, a Go channel that signals when new entries are ready to be reported. The second is the lastApplied variable, which tracks which log entries have already been handed to the service through the commit channel. Entries are only forwarded after they cross the commit threshold.
Leader's Replication Loop
A new command at the leader doesn't reach followers by itself. The leader pushes entries out during its periodic heartbeat tick. The leaderSendHeartbeats method, which runs on every leader tick, is where the AppendEntries (AE) RPC is now fully populated with prevLogIndex, prevLogTerm, log entries, and leaderCommit. The response contains a success bit that tells the leader whether the follower's log matched the expected prefix; if not, the leader steps nextIndex back and retries on the next tick.
The leader also adjusts its commitIndex based on follower acknowledgements. The majority rule applies: once a log index is replicated on a majority of servers, the leader advances commitIndex to that index. The updated commitment is broadcast to followers in the leaderCommit field of later AE RPCs.
func (cm *ConsensusModule) leaderSendHeartbeats() {
cm.mu.Lock()
savedCurrentTerm := cm.currentTerm
cm.mu.Unlock()
for _, peerId := range cm.peerIds {
go func(peerId int) {
cm.mu.Lock()
ni := cm.nextIndex[peerId]
prevLogIndex := ni - 1
prevLogTerm := -1
if prevLogIndex >= 0 {
prevLogTerm = cm.log[prevLogIndex].Term
}
entries := cm.log[ni:]
args := AppendEntriesArgs{
Term: savedCurrentTerm,
LeaderId: cm.id,
PrevLogIndex: prevLogIndex,
PrevLogTerm: prevLogTerm,
Entries: entries,
LeaderCommit: cm.commitIndex,
}
cm.mu.Unlock()
cm.dlog("sending AppendEntries to %v: ni=%d, args=%+v", peerId, ni, args)
var reply AppendEntriesReply
if err := cm.server.Call(peerId, "ConsensusModule.AppendEntries", args, &reply); err == nil {
cm.mu.Lock()
defer cm.mu.Unlock()
if reply.Term > savedCurrentTerm {
cm.dlog("term out of date in heartbeat reply")
cm.becomeFollower(reply.Term)
return
}
if cm.state == Leader && savedCurrentTerm == reply.Term {
if reply.Success {
cm.nextIndex[peerId] = ni + len(entries)
cm.matchIndex[peerId] = cm.nextIndex[peerId] - 1
cm.dlog("AppendEntries reply from %d success: nextIndex := %v, matchIndex := %v", peerId, cm.nextIndex, cm.matchIndex)
savedCommitIndex := cm.commitIndex
for i := cm.commitIndex + 1; i < len(cm.log); i++ {
if cm.log[i].Term == cm.currentTerm {
matchCount := 1
for _, peerId := range cm.peerIds {
if cm.matchIndex[peerId] >= i {
matchCount++
}
}
if matchCount*2 > len(cm.peerIds)+1 {
cm.commitIndex = i
}
}
}
if cm.commitIndex != savedCommitIndex {
cm.dlog("leader sets commitIndex := %d", cm.commitIndex)
cm.newCommitReadyChan <- struct{}{}
}
} else {
cm.nextIndex[peerId] = ni - 1
cm.dlog("AppendEntries reply from %d !success: nextIndex := %d", peerId, ni-1)
}
}
}
}(peerId)
}
}
if cm.commitIndex != savedCommitIndex {
cm.dlog("leader sets commitIndex := %d", cm.commitIndex)
cm.newCommitReadyChan <- struct{}{}
}
func (cm *ConsensusModule) commitChanSender() {
for range cm.newCommitReadyChan {
// Find which entries we have to apply.
cm.mu.Lock()
savedTerm := cm.currentTerm
savedLastApplied := cm.lastApplied
var entries []LogEntry
if cm.commitIndex > cm.lastApplied {
entries = cm.log[cm.lastApplied+1 : cm.commitIndex+1]
cm.lastApplied = cm.commitIndex
}
cm.mu.Unlock()
cm.dlog("commitChanSender entries=%v, savedLastApplied=%d", entries, savedLastApplied)
for i, entry := range entries {
cm.commitChan <- CommitEntry{
Command: entry.Command,
Index: savedLastApplied + i + 1,
Term: savedTerm,
}
}
}
cm.dlog("commitChanSender done")
}
Follower-Side Updates and Commitment
Followers receive these AE RPCs and apply the updates to their own logs. The handler first checks that prevLogIndex and prevLogTerm match an existing log entry; if not, it responds with success=false. On a match, it appends the incoming entries (truncating any conflicting suffix first) and checks whether the leader has told it about a higher commit index. If leaderCommit exceeds the follower's own commitIndex, the follower advances its commit point and signals the internal ready channel so that new committed entries get forwarded to the service.
Two full RPC round trips are required to commit a fresh command. The first trip carries the log entries to followers and returns their acknowledgements. Only after the leader counts a majority does it advance its commit index, which is then communicated back to followers in the next AE round. Those followers, upon seeing the higher leaderCommit, mark the new entries as committed and report them on the commit channel.
Election Safety
Log replication changes the rules for who may become leader. Raft's election restriction (paper section 5.4.1) prevents a candidate from winning unless its log is at least as up-to-date as a majority of peers. To enforce this, RequestVote (RV) requests now carry the candidate's lastLogIndex and lastLogTerm, taken from the candidate's final log entry.
func (cm *ConsensusModule) startElection() {
cm.state = Candidate
cm.currentTerm += 1
savedCurrentTerm := cm.currentTerm
cm.electionResetEvent = time.Now()
cm.votedFor = cm.id
cm.dlog("becomes Candidate (currentTerm=%d); log=%v", savedCurrentTerm, cm.log)
votesReceived := 1
// Send RequestVote RPCs to all other servers concurrently.
for _, peerId := range cm.peerIds {
go func(peerId int) {
cm.mu.Lock()
savedLastLogIndex, savedLastLogTerm := cm.lastLogIndexAndTerm()
cm.mu.Unlock()
args := RequestVoteArgs{
Term: savedCurrentTerm,
CandidateId: cm.id,
LastLogIndex: savedLastLogIndex,
LastLogTerm: savedLastLogTerm,
}
cm.dlog("sending RequestVote to %d: %+v", peerId, args)
var reply RequestVoteReply
if err := cm.server.Call(peerId, "ConsensusModule.RequestVote", args, &reply); err == nil {
cm.mu.Lock()
defer cm.mu.Unlock()
cm.dlog("received RequestVoteReply %+v", reply)
if cm.state != Candidate {
cm.dlog("while waiting for reply, state = %v", cm.state)
return
}
if reply.Term > savedCurrentTerm {
cm.dlog("term out of date in RequestVoteReply")
cm.becomeFollower(reply.Term)
return
} else if reply.Term == savedCurrentTerm {
if reply.VoteGranted {
votesReceived++
if votesReceived*2 > len(cm.peerIds)+1 {
// Won the election!
cm.dlog("wins election with %d votes", votes)
cm.startLeader()
return
}
}
}
}
}(peerId)
}
// Run another election timer, in case this election is not successful.
go cm.runElectionTimer()
}
The lastLogIndexAndTerm helper returns these values, with a sentinel of -1 for an empty log (the implementation uses 0-based indexing, unlike the paper's 1-based scheme). When a follower receives an RV request, it compares the candidate's last log term and index against its own. Only a candidate whose log is equally fresh or newer gets the vote.
// lastLogIndexAndTerm returns the last log index and the last log entry's term
// (or -1 if there's no log) for this server.
// Expects cm.mu to be locked.
func (cm *ConsensusModule) lastLogIndexAndTerm() (int, int) {
if len(cm.log) > 0 {
lastIndex := len(cm.log) - 1
return lastIndex, cm.log[lastIndex].Term
} else {
return -1, -1
}
}
func (cm *ConsensusModule) RequestVote(args RequestVoteArgs, reply *RequestVoteReply) error {
cm.mu.Lock()
defer cm.mu.Unlock()
if cm.state == Dead {
return nil
}
lastLogIndex, lastLogTerm := cm.lastLogIndexAndTerm()
cm.dlog("RequestVote: %+v [currentTerm=%d, votedFor=%d, log index/term=(%d, %d)]", args, cm.currentTerm, cm.votedFor, lastLogIndex, lastLogTerm)
if args.Term > cm.currentTerm {
cm.dlog("... term out of date in RequestVote")
cm.becomeFollower(args.Term)
}
if cm.currentTerm == args.Term &&
(cm.votedFor == -1 || cm.votedFor == args.CandidateId) &&
(args.LastLogTerm > lastLogTerm ||
(args.LastLogTerm == lastLogTerm && args.LastLogIndex >= lastLogIndex)) {
reply.VoteGranted = true
cm.votedFor = args.CandidateId
cm.electionResetEvent = time.Now()
} else {
reply.VoteGranted = false
}
reply.Term = cm.currentTerm
cm.dlog("... RequestVote reply: %+v", reply)
return nil
}
Rejoining a Disconnected Server
In Part 1, a disconnected server B that reconnects after running repeated elections would return with an inflated term and trigger a new election. Now suppose that during B's absence the two connected peers replicated several new entries. When B rejoins, the election safety check steps in. Because B's log is necessarily behind A's and C's — it missed every entry committed while it was disconnected — B cannot win the vote. One of the connected peers with a fuller log takes leadership, keeping disruption minimal.
An unnecessary re-election still occurs in this scenario. Ongaro's dissertation discusses this problem under "Preventing disruptions when a server rejoins a cluster"; the common fix is a pre-vote phase that filters out lagging candidates before they disturb the cluster. Since that is an optimization for an uncommon edge case, the implementation here does not include it.
Why commitIndex and lastApplied Are Separate
A common question when implementing Raft is why the leader and followers maintain two separate indices: commitIndex and lastApplied. The distinction exists to decouple fast operations, like handling RPCs, from potentially slow ones, such as delivering committed commands to clients over a channel or callback.
When a follower receives an AppendEntries RPC and learns the leader's commitIndex has advanced, it may need to forward a batch of log entries to the commit channel. That send can block if the client isn't reading promptly, which would delay the RPC reply. By only updating commitIndex inside the RPC handler and letting a background goroutine (commitChanSender) observe changes and push commands to the client at its own pace, the RPC path stays fast and non-blocking.
The same logic applies to the newCommitReadyChan internal notification channel. That channel is buffered, and because both the producer and consumer are within our control, a small buffer suffices to avoid blocking in nearly all cases. However, an extremely slow client could still cause backpressure on RPC handling — this is acceptable, as it provides a natural flow-control mechanism.
Why Leaders Track Both nextIndex and matchIndex
Another frequent question is whether a leader really needs per-peer nextIndex and matchIndex values. Technically, the algorithm would function with only matchIndex, but at a significant performance cost in common scenarios.
Consider a fresh leadership election. A new leader has no knowledge of its followers' log state, so it would set matchIndex to -1 and attempt to send its entire log to every peer. In practice, most followers already have nearly all of those entries. Maintaining nextIndex lets the leader probe each follower starting from the presumed end of its log, avoiding the unnecessary transfer of large logs and making log catch-up far more efficient.
Current Limitations and Next Steps
The current implementation is functional for basic operation, but it lacks persistence. This means the system is vulnerable to crash faults: if a server restarts after a crash, it loses its volatile state and cannot participate correctly until it re-syncs. Raft's design accounts for this, and adding persistence will allow for more demanding tests where servers fail at the most inconvenient moments.
There is also room for optimization in the current design. Leaders currently send AppendEntries RPCs on a fixed 50 ms interval, even when they have new information to deliver to followers. In the next part, this promptness issue will be addressed so that followers receive updates more quickly when the leader's log advances.
You can explore the code and run tests with logging enabled to observe the behavior firsthand.
