Raft: Log Replication¶
The job¶
Once a leader exists, it accepts commands, appends them to its log,
replicates them to followers, decides when each entry is committed
(safe on a majority, never to be lost), and every node applies
committed entries to its state machine in log order. The guarantee
Raft gives the state machine is State Machine Safety: no two nodes
ever apply different commands at the same log index. This page is the
replication half of labs/raft; elections are in
Raft: Leader Election.
Proposing a command¶
// Start proposes a command. It returns immediately; the command is
// committed only if it later appears on the apply channel at this index
// (a leader can lose leadership before replicating it).
func (n *Node) Start(cmd []byte) (index, term uint64, isLeader bool) {
n.mu.Lock()
defer n.mu.Unlock()
if n.state != Leader || n.dead.Load() {
return 0, n.term, false
}
e := n.appendLocal(cmd) // append + persist
n.broadcastLocked() // replicate now instead of waiting for the next heartbeat
return e.Index, e.Term, true
}
Start returning an index is a hope, not a promise. If this leader
is deposed before the entry reaches a majority, a different command may
be committed at that index. The caller must watch the apply channel and
check that the command applied at index is its own — the KV server
does exactly that (Replicated Key-Value Store).
Sending entries¶
Simplified here to the core logic; the current code also sends a snapshot when the follower needs entries that were compacted away, and keeps at most one RPC in flight per follower (see Raft in Production).
func (n *Node) replicateTo(p int) {
n.mu.Lock()
if n.state != Leader {
n.mu.Unlock()
return
}
next := n.nextIndex[p]
prev := n.log[next-1]
args := AppendEntriesArgs{
Term: n.term, LeaderID: n.id,
PrevLogIndex: prev.Index, PrevLogTerm: prev.Term,
Entries: append([]Entry(nil), n.log[next:]...), // copy: the log may change after unlock
LeaderCommit: n.commitIndex,
}
n.mu.Unlock()
reply, ok := n.tr.AppendEntries(p, args)
if !ok {
return // the next heartbeat retries
}
n.mu.Lock()
defer n.mu.Unlock()
if reply.Term > n.term {
n.becomeFollower(reply.Term)
return
}
if n.state != Leader || n.term != args.Term {
return // stale reply from an earlier term
}
if reply.Success {
match := args.PrevLogIndex + uint64(len(args.Entries))
if match > n.matchIndex[p] { // replies can arrive out of order: never move backwards
n.matchIndex[p] = match
n.nextIndex[p] = match + 1
n.advanceCommitLocked()
}
return
}
// Mismatch: jump back past the follower's conflicting term.
next = reply.ConflictIndex
if reply.ConflictTerm != 0 {
for i := n.lastIndex(); i > 0; i-- {
if n.log[i].Term == reply.ConflictTerm {
next = i + 1
break
}
}
}
n.nextIndex[p] = max(1, min(next, n.lastIndex()+1))
}
Two subtleties:
matchis computed from the arguments that were sent, not from the leader's current log ornextIndex, both of which may have changed while the RPC was in flight.matchIndexonly moves forward: a delayed success reply for an older, shorter request must not undo a newer one.
Receiving entries: the consistency check¶
func (n *Node) HandleAppendEntries(args AppendEntriesArgs) AppendEntriesReply {
n.mu.Lock()
defer n.mu.Unlock()
if args.Term < n.term {
return AppendEntriesReply{Term: n.term} // a deposed leader: tell it the new term
}
n.becomeFollower(args.Term)
n.leaderID = args.LeaderID
n.resetElectionTimer()
// Our log must contain the entry preceding the new ones.
if args.PrevLogIndex > n.lastIndex() {
return AppendEntriesReply{Term: n.term, ConflictIndex: n.lastIndex() + 1}
}
if t := n.log[args.PrevLogIndex].Term; t != args.PrevLogTerm {
first := args.PrevLogIndex
for first > 1 && n.log[first-1].Term == t {
first--
}
return AppendEntriesReply{Term: n.term, ConflictTerm: t, ConflictIndex: first}
}
// Append, truncating only at the first real conflict.
changed := false
for i, e := range args.Entries {
idx := args.PrevLogIndex + 1 + uint64(i)
if idx <= n.lastIndex() {
if n.log[idx].Term == e.Term {
continue // already have it
}
n.log = n.log[:idx] // conflict: drop this entry and everything after
}
n.log = append(n.log, args.Entries[i:]...)
changed = true
break
}
if changed {
n.save(Update{HardState: n.hardState(), Entries: args.Entries[i:]}) // durable before we acknowledge
}
if args.LeaderCommit > n.commitIndex {
// Never move commitIndex backwards (a delayed heartbeat may carry an old prefix).
n.commitIndex = max(n.commitIndex,
min(args.LeaderCommit, args.PrevLogIndex+uint64(len(args.Entries))))
n.applyCond.Broadcast()
}
return AppendEntriesReply{Term: n.term, Success: true}
}
The Log Matching Property follows by induction: if two logs contain an entry with the same index and term, the logs are identical up to that index — because a leader creates at most one entry per index per term, and a follower accepts entries only directly after an entry that matches the leader's.
The truncation rule is the most common source of bugs: a follower must not truncate its log just because an AppendEntries is shorter than its log. With a network that delays and reorders messages, an old request carrying entries 5–7 can arrive after a newer one carrying 5–10; truncating to 7 would delete entries the follower already acknowledged — and that a leader may already have counted toward a commit.
Fast backtracking¶
When the follower's log diverges, the leader must find the last index
where they agree. Decrementing nextIndex by one per rejected RPC costs
one round trip per conflicting entry. The follower instead reports the
term of its conflicting entry and the first index of that term,
so the leader skips a whole term per round trip.
Measured in TestDivergentLogsRepaired: a 5-node cluster is
partitioned with the leader in a 2-node minority; the old leader
appends 50 entries that can never commit, while the majority commits 50
different entries under a new leader. After the partition heals, the
cluster repairs both divergent logs (the old leader's and its
follower's) and commits a new entry on all 5 nodes using 12 RPCs in
total — including heartbeats to the three nodes that were already up to
date. One-by-one backtracking would have needed about 50 rejected round
trips per divergent node.
When is an entry committed?¶
An entry is committed once it is stored on a majority — with one exception that breaks naive implementations. A leader may only count replicas for entries of its current term:
func (n *Node) advanceCommitLocked() {
for idx := n.lastIndex(); idx > n.commitIndex; idx-- {
if n.log[idx].Term != n.term {
break // older-term entries commit only indirectly
}
count := 0
for _, p := range n.peers {
if n.matchIndex[p] >= idx {
count++
}
}
if count >= n.majority() {
n.commitIndex = idx
n.applyCond.Broadcast()
n.broadcastLocked() // tell followers the new commit index promptly
return
}
}
}
Why (the Raft paper's Figure 8): an entry from an old term can be on a majority and still be overwritten, because a node that never received it can be elected in a later term with a log whose last term is newer. Only once an entry from the leader's own term is on a majority is every earlier entry protected by the election restriction. Committing that entry commits everything before it.
This is why becomeLeader appends a no-op entry: a new leader with
uncommitted entries from earlier terms cannot commit them until it
commits something from its own term. Without the no-op, a quiet cluster
would leave those entries uncommitted until the next client write — and
ReadIndex could not know its commit index is current.
Applying¶
func (n *Node) applier() {
defer close(n.applyCh)
for {
n.mu.Lock()
for n.lastApplied >= n.commitIndex && !n.dead.Load() {
n.applyCond.Wait()
}
if n.dead.Load() {
n.mu.Unlock()
return
}
var msgs []ApplyMsg
for i := n.lastApplied + 1; i <= n.commitIndex; i++ {
e := n.log[i]
msgs = append(msgs, ApplyMsg{Index: e.Index, Term: e.Term, Command: e.Command})
}
n.lastApplied = n.commitIndex
n.mu.Unlock()
for _, m := range msgs { // send WITHOUT holding mu
select {
case n.applyCh <- m:
case <-n.done:
return
}
}
}
}
A dedicated goroutine delivers committed entries in order. Sending on
the channel while holding mu would deadlock as soon as the consumer
calls back into the node (for example, Start from the KV server) or
simply falls behind.
Persistence¶
Storage is an interface that receives changes, not whole states:
type Update struct {
HardState HardState // term and vote
Entries []Entry // replace the log from Entries[0].Index onward
Snapshot *Snapshot // entries up to Snapshot.Index are no longer needed
DiscardLog bool // the follower's log conflicted with the snapshot
}
type Storage interface {
Load() (HardState, *Snapshot, []Entry, error)
Save(Update) error // durable before returning
}
Every place that changes persistent state calls n.save(...) with just
what changed — a vote is a hard-state update, a follower's append is a
log suffix, a leader's proposal is one or more new entries. Two
implementations exist: MemStorage for tests (a restarted node is handed
the same object, simulating storage that survived the crash), and
labs/raftwal, an append-only write-ahead log on disk. On real
hardware the WAL makes each durable save cost one small append and one
fsync, independent of log size; rewriting the whole state on every save
costs more and grows with the log — measured in
Raft in Production.
Safety under chaos¶
TestChaos runs 5 nodes over an unreliable simulated network (10%
of messages lost, 1–10 ms delays, reordering), and for 200 rounds
randomly crashes nodes (restarting them from persisted state), creates
and heals partitions, and submits commands on every node. A consumer
for each node checks, on every applied entry, that it arrives in order
with no gaps and that no other node ever applied a different command
at the same index. At the end the network is healed and one more
command must reach all 5 nodes. It passes for 5 seeds; testing is
covered in Testing Distributed Systems.
What production Raft adds¶
Snapshots and InstallSnapshot, membership changes, flow control and
batching, and durable WAL storage are now implemented in labs/raft
and labs/raftwal; they are covered in
Raft in Production. What remains outside this
implementation: joint consensus (changing several members at once),
learners, leadership transfer, and writing the leader's log to disk in
parallel with sending it to followers.
Common pitfall¶
Committing an old-term entry because a majority stores it. An
implementation counts matchIndex for every index above commitIndex
regardless of term. Under the right sequence — leader S1 replicates an
entry from term 2 to a majority while running in term 4, S1 crashes, S5
(which never had the entry) is elected with a term-3 entry at that index
— S5 overwrites the "committed" entry on every node. Clients were told
their write succeeded; it is gone. The bug needs crashes and elections
in a specific order, so ordinary tests rarely hit it; randomized tests
with a State Machine Safety check across nodes, run with many seeds,
find it. The fix is the single if n.log[idx].Term != n.term line above,
plus the no-op entry so the leader can make progress.
Review questions¶
- Why must the leader compute
matchfrom the request arguments rather than fromnextIndexafter the reply arrives? - Why must a follower not truncate its log when an AppendEntries is shorter than its log?
- How do
ConflictTermandConflictIndexlet the leader skip a whole term per round trip? - Explain why a leader may not commit an old-term entry just because a majority has it. What does the no-op entry fix?
- Why does the applier send on the channel without holding the node's mutex?