A6 · Part 3 · Advanced
Distributed consensus (surface)
Getting a group of unreliable machines to agree on one value — and why your job is to use a consensus system correctly, not to write one.
10On this page
- What consensus buys, and where you actually need it
- The problem, stated honestly
- Raft: terms, elections, and why randomization matters
- Log replication and the commit rule
- Cluster sizing — always odd, almost always 3 or 5
- Consensus and its cousins
- Paxos, briefly
- In practice: coordination services, split brain, and fencing
- Why you should almost never implement consensus yourself
- Recognizing it in an unseen problem
What consensus buys, and where you actually need it
Consensus is the problem of getting a set of nodes to agree on a single value, such that once a value is decided, every node that ever learns a decision learns the same one — even though nodes crash and restart, and messages are delayed, reordered, duplicated, or lost. It sounds abstract until you notice how many concrete production problems are secretly the same problem.
| Problem | The value being agreed on |
|---|---|
| Leader election | "Which node is the primary for term 7?" Every replication scheme with a single writer needs this, and needs it to be unambiguous. |
| Cluster membership and config | "Which nodes are in the cluster, and what is the current shard map?" A split view of membership is how you get two nodes both believing they own shard 12. |
| Distributed locks / leases | "Who holds the right to run the nightly billing job?" Running it twice is a customer-visible incident. |
| Atomic commit across shards | "Did transaction T commit or abort?" All participants must reach the same answer. |
| Exactly-once semantics | "Has message 91,442 been processed?" Exactly-once delivery is impossible; exactly-once effect is achievable by agreeing on a durable sequence and deduplicating against it. |
Equally important is the list of things that do not need consensus, because reaching for it unnecessarily is its own mistake: ordinary application data (use replication with quorums), counters and metrics (use CRDTs or just accept approximation), caches, and anything where "two nodes did the work" is merely wasteful rather than wrong. Consensus costs a majority round trip on every decision — you pay it only where correctness demands it.
The problem, stated honestly
What makes this hard is that in an asynchronous network you cannot distinguish a crashed node from a slow node from a node you can't currently reach. All three look identical: silence. The FLP result formalises this — with even one possible crash and no timing assumptions, no deterministic protocol can guarantee termination. Real systems escape it by adding a timing assumption in the form of timeouts, which makes progress probabilistic while keeping safety absolute. That split is the key idea to hold onto:
- Safety — never two different decisions, never a lost committed entry — holds unconditionally, even during arbitrary partitions and crashes.
- Liveness — actually deciding something — holds only when a majority can talk to each other for long enough. During a bad partition, a consensus system stops making progress. That is not a bug; it is the CP choice being exercised.
Raft: terms, elections, and why randomization matters
Raft was explicitly designed to be understandable, which is why it is the one to explain out loud. It decomposes consensus into leader election, log replication, and safety.
Time is divided into terms: monotonically increasing integers that act as a logical clock. Each term has at most one leader. Every message carries a term; a node seeing a higher term immediately steps down to follower and adopts it, and a node seeing a lower term rejects the message. That single rule retires stale leaders automatically.
A follower that hears no heartbeat within its election timeout becomes a candidate: it increments the term, votes for itself, and asks everyone for a vote. A node grants at most one vote per term, and only to a candidate whose log is at least as up to date as its own. Win a majority and you are leader; the first thing a leader does is send heartbeats to suppress further elections.
Why a majority guarantees at most one leader: two majorities of the same cluster must intersect in at least one node, and that node casts at most one vote per term. So two candidates cannot both reach a majority in the same term, and higher terms retire lower ones. This is the sentence to be able to say verbatim.
Why the timeout is randomized: if every follower used the same election timeout, they would all time out together, all become candidates, all split the vote, and all time out again — livelock. Randomizing each node's timeout over a range (the Raft paper suggests 150–300 ms; etcd defaults to 1000 ms with 100 ms heartbeats) means one node almost always wakes first and wins before the others start. Randomization is not a tuning detail; it is the liveness mechanism.
Dry run: a five-node election where two candidates collide and one still wins
Assume one-way message delays of 1–2 ms, with n2 and n3 close to n1 and n4 close to n5. This matches the figure above.
| Time (ms) | Event | Votes granted to n1 | Votes granted to n5 | Cluster state |
|---|---|---|---|---|
| 0 | Leader's heartbeats stop arriving | 0 | 0 | Term 4, all five nodes idle followers |
| 147 | n1's randomized timeout fires first; it becomes a candidate, increments to term 5, votes for itself | 1 (self) | 0 | n1 sends RequestVote(term 5) to n2, n3, n4, n5 |
| 148 | n5's timeout fires 1 ms later, before n1's request can reach it; it also becomes a candidate in term 5 and votes for itself | 1 | 1 (self) | n5 sends RequestVote(term 5) to n1, n2, n3, n4 |
| 149 | n2 and n3 receive n1's request first and grant it; n4 receives n5's request first and grants that | 3 | 2 | Each voter has now used its single term-5 vote |
| 150 | The second requests arrive: n4 rejects n1, n2 and n3 reject n5, and n1 and n5 reject each other (each already voted for itself) | 3 | 2 | Nobody can gain another vote in term 5 |
| 151 | The grants from n2 and n3 reach n1; it now holds 3 of 5, a majority | 3 | 2 | n1 becomes leader of term 5 and sends heartbeats at once; n5 stays stuck at 2 of 5 |
| ~153 | n1's first heartbeat (AppendEntries, term 5) reaches n5 | 3 | 2 | n5 is a candidate in term 5 and sees a leader for term 5, so it steps down to follower — no second election |
With two candidates among five nodes, when all three other voters respond they cannot split evenly, so one candidate reaches three votes. Randomization earns its keep when three or more nodes time out together (votes can split 2-2-1 and nobody wins): without randomization they would retry on the same schedule and collide again and again, while with it one node almost always wakes first. That is why it is called the liveness mechanism, not a tuning knob. If n1 and n5 had fired in the same millisecond the table would look the same, with arrival order at each voter deciding the 3-2 split.
PreVote and CheckQuorum: keeping a rejoining node from disrupting the cluster
Plain Raft has a nasty edge. A node cut off from the majority keeps timing
out, becoming a candidate, and incrementing its term, so after a long
partition it holds a very high term. When it rejoins, its first message
makes the healthy leader see a higher term and step down, forcing an
election the cluster did not need. Two refinements from the Raft
dissertation fix it. PreVote: before bumping its term, a would-be
candidate asks peers whether they would vote for it; peers that still hear
from a live leader say no, so a minority node never inflates its term.
CheckQuorum: a leader that has not heard from a majority within an
election timeout steps down on its own. A related rule in the dissertation
has nodes that recently heard from a leader ignore vote requests. Production implementations enable them
(etcd's --pre-vote flag defaults to true in 3.6), and this
is why a rejoining minority node is usually a non-event.
Log replication and the commit rule
Every state change is an entry appended to a replicated log. Clients send
commands to the leader; the leader appends locally, then sends
AppendEntries to followers. An entry is committed once
it is durably stored on a majority — at which point the leader applies it
to its state machine, returns to the client, and tells followers the new
commit index on the next heartbeat. Because every replica applies the same
entries in the same order, every replica ends in the same state. That is
the replicated state machine pattern, and it is what a consensus
system really sells you.
| Raft property | What it prevents |
|---|---|
Log matching: AppendEntries carries the index and term of the preceding entry; a follower rejects it if that doesn't match, and the leader walks backwards until it finds agreement, then overwrites the divergent tail. | Divergent histories silently persisting on a follower. |
| Election restriction: a voter refuses any candidate whose last log entry is older (lower term, or same term but shorter) than its own. | Electing a leader that is missing a committed entry — which would erase it. |
| Leader-term commit rule: a leader may only mark an entry committed by counting replicas if that entry is from its own term. Entries from previous terms become committed indirectly, once a current-term entry above them commits. | The subtle Figure-8 scenario where an entry replicated to a majority is still later overwritten. Naming this rule is a genuine expert signal. |
Persist before responding: currentTerm, votedFor, and the log must be fsynced before any reply. | A node that crashes and restarts voting twice in the same term — which elects two leaders and destroys safety. |
⚠ Reads are not automatically safe
A leader that has been silently partitioned away still believes
it is leader until its next failed heartbeat round. If it answers reads
from local state in that window, it serves stale data — a real
linearizability violation. Real systems fix this with a read-index (confirm
leadership with a heartbeat round before answering) or a leader lease
(answer locally, but only within a lease shorter than the election
timeout). ZooKeeper makes a different trade: any server answers reads
from its own local state, so reads can be stale by design and you call
sync() first when you need the latest committed write. That
is a separate issue from the stale-leader problem above.
Cluster sizing — always odd, almost always 3 or 5
| Nodes | Majority | Failures tolerated | Comment |
|---|---|---|---|
| 3 | 2 | 1 | The default. Survives one node or one AZ. |
| 4 | 3 | 1 | Strictly worse than 3 — same fault tolerance, more nodes to ack every write. Avoid it. |
| 5 | 3 | 2 | The right answer for anything critical: survives a node failure during a maintenance window. |
| 7 | 4 | 3 | Rarely worth it. Every write waits for the 4th-fastest node, so latency gets worse as you add members. |
Consensus clusters do not scale by adding members — throughput drops as they grow, because every decision needs a majority ack. You scale them by sharding the keyspace across many independent consensus groups, which is exactly what Spanner (one Paxos group per range) and CockroachDB (one Raft group per range) do. As a planning figure, expect writes of a few milliseconds for a same-region etcd cluster (dominated by the disk fsync), and a full inter-region round trip, 60 ms or more across the US, if you stretch the members across regions.
Consensus and its cousins
Three equivalences are worth one sentence each. Total order broadcast (every node delivers the same messages in the same order) is equivalent to consensus: a replicated log is repeated consensus. A linearizable compare-and-set register is too, which is why leader election and locks reduce to it. Two-phase commit is not consensus: it is atomic commit with a single coordinator, and if that coordinator dies after participants vote yes, they block holding locks. Spanner and CockroachDB run two-phase commit across consensus groups, so each participant and the coordinator state are themselves replicated. And Raft and Multi-Paxos tolerate crash faults only; nodes that lie or are compromised need a Byzantine-fault-tolerant protocol with 3f+1 replicas, which is a different (and rarely needed) problem. Related variants you may meet: non-voting learner members that catch up before being promoted, leader leases for local reads, and Flexible Paxos and EPaxos, which relax the "one majority on every decision" rule.
Paxos, briefly
Paxos is the original (Lamport, 1989/1998) and is still the substrate for Chubby and Spanner. Single-decree Paxos agrees on one value with prepare and accept phases; Multi-Paxos amortises the prepare phase by keeping a stable leader — at which point it looks very much like Raft. It is notoriously hard to specify completely enough to implement, which is precisely the gap Raft was written to fill. In an interview: mention it as "the older, harder one that Raft was designed to replace as the teachable protocol", note that Multi-Paxos and Raft are equivalent in power, and move on. Do not attempt to derive Paxos at a whiteboard.
In practice: coordination services, split brain, and fencing
You almost never talk to Raft directly. You talk to a coordination service that has already solved it and exposes a small, safe API.
| Service | Protocol | Shape of the API | Reach for it when… |
|---|---|---|---|
| etcd | Raft | Key-value with revisions, leases, compare-and-swap, watches. Every write returns a monotonically increasing revision. | Kubernetes-adjacent infrastructure, service discovery, leader election, config. The default modern choice. |
| ZooKeeper | Zab | Hierarchical znodes, ephemeral and sequential nodes, watches. Ephemeral+sequential is the classic leader-election recipe. | The JVM ecosystem — HBase, Solr, and Kafka before 4.0 (which is KRaft-only). Battle-tested for well over a decade. |
| Consul | Raft | KV, sessions, health checks, service catalogue, DNS interface. | Service discovery where health checking and multi-datacenter federation matter as much as the KV store. |
| Your database | varies | A row with a unique constraint plus a lease column is a perfectly good lock if you already have a strongly consistent database. | You don't want another stateful system to operate. Often the correct, boring answer. |
Split brain is what happens when two nodes simultaneously believe they are the leader. Majority quorum prevents two leaders from both committing, but it does not prevent an old leader from thinking it is leader and acting on external systems that have no idea a quorum exists. The lock service is consistent; your S3 bucket, your payment gateway, and your file system are not participants in the protocol.
The fix is a fencing token: the lock service hands out a monotonically increasing number with every grant, and every write to the protected resource carries it. The resource remembers the highest token it has seen and rejects anything lower. This moves the arbitration to the resource, which is the only place it can be correct.
You usually get the token for free: ZooKeeper's zxid or znode
version, etcd's revision of the lock key (its create or mod revision; do
not use the lease ID, which does not record the order of writes), or a version column you compare-and-swap on in your
own database. Kleppmann's essay on distributed locking is the standard
reference for why the token is needed and why Redis-based locks lack one.
This is also the honest answer to the long-running argument about
Redis-based distributed locks: a multi-node Redis lock relies on bounded
clock drift and bounded process pauses, and neither is guaranteed — so if
you propose one, be ready to say what happens during a 30-second GC pause,
and carry a fencing token so the answer is "nothing bad".
Say it like this → "I'll use an etcd lease for
the leader lock, but I won't trust the lock on its own — the leader
attaches the lock key's etcd revision as a fencing token to every write, and the
storage layer rejects any write whose token is lower than the highest it
has seen. That way a leader that was partitioned away and doesn't know it
yet can't corrupt anything when it comes back."
Why you should almost never implement consensus yourself
The happy path of Raft is a weekend project. The parts that take years are the ones that only show up under failure:
- Persistence correctness.
currentTermandvotedFormust be fsynced before you reply, or a crash-restart lets a node vote twice in one term and you elect two leaders. This bug is invisible in testing and unrecoverable in production. - Membership changes. Adding or removing a node while the cluster is live requires joint consensus or strict single-node changes; get it wrong and you create two disjoint majorities.
- Log compaction and snapshots. Logs grow forever otherwise, and installing a snapshot on a lagging follower is a whole second protocol.
- The read path. As above — the naive implementation quietly serves stale reads.
- Verification. Production implementations are TLA+-specified and tested with deterministic fault injection (Jepsen, FoundationDB-style simulation). Consensus bugs are silent, and they lose committed data.
The correct engineering answer is: use etcd, ZooKeeper, or Consul; or use a database that already embeds a verified implementation (Spanner, CockroachDB, YugabyteDB, TiDB, Kafka's KRaft). Better still, design so you need consensus in as few places as possible — one small, well-understood coordination layer that everything else defers to.
⚠ The consensus-shaped answer that isn't
"I'll have the nodes vote among themselves" and "I'll use a heartbeat and
whoever stops responding gets replaced" both describe a system with no
quorum requirement — that is a split-brain generator, not consensus. If
you cannot say which set of nodes must agree and why two such sets must
overlap, you have not described a consensus protocol.
Recognizing it in an unseen problem
- The prompt says "exactly one", "only one node should", "elect", "must not run twice", or "who owns this partition" — those are all leader election in disguise.
- Any design with a single writer, a primary, or a coordinator has an implicit consensus dependency: ask yourself who decides who the primary is, and what happens if two nodes disagree.
- A naive design says "we'll use a lock in Redis." The follow-up is always the pause scenario. Have the fencing token ready before it's asked — it converts a shaky answer into a senior one.
- Distinguish from quorum replication: quorums give you consistency for a single key; consensus gives you an agreed ordered log, which is what you need for leadership, membership, and atomic commit.
- Distinguish from CAP: consensus systems are the CP corner made concrete — during a partition the minority stops. If the prompt cannot tolerate stopping, you need to move that data out of the consensus path entirely.
- Scope it explicitly. A good answer keeps consensus on a small amount of metadata (leases, shard maps, config) and keeps the bulk data out of it, because a consensus group's throughput is bounded by a majority round trip.
Before you move on
- Explain why a majority guarantees at most one leader per term, using the set-overlap argument.
- Walk through the five-node election dry run above and say what would change if n1 and n5 had timed out at the exact same millisecond.
- Explain why randomized election timeouts are a liveness mechanism, not a tuning detail.
- Describe what a fencing token fixes that a lock service alone cannot, and trace it through the client-A / client-B scenario in this chapter.
- Say, for a design you're sketching, exactly which piece of state needs consensus and which pieces don't — and why reaching for etcd or ZooKeeper is usually right instead of writing Raft yourself.
- Explain what PreVote prevents and why a long-partitioned minority node can otherwise unseat a healthy leader when it rejoins.
Sources and further reading
- In search of an understandable consensus algorithm (the Raft paper) — terms, elections, log matching, the commit rule and the 150-300 ms timeout suggestion
- Ongaro's Raft dissertation — PreVote, CheckQuorum, membership changes and read-only queries
- The Raft website — visualisation and a list of implementations
- Kleppmann, how to do distributed locking — fencing tokens, and why Redis-based locks lack them
- etcd API — revisions, create and mod revision, and what a lease ID is
- etcd configuration flags — pre-vote default, heartbeat and election timeout defaults
- etcd API guarantees — linearizable reads by default, strict serializability, opt-in stale reads
- ZooKeeper programmer's guide — local reads can be stale; sync before reading when it matters
- Spanner paper (OSDI 2012) — a Paxos group per tablet with two-phase commit across groups
Claims in this chapter were checked against these sources in October 2026. Figures marked "planning figure" are rules of thumb, not guarantees.
5 questions drawn from 7, scored out of 5. This is practice: it runs in your browser, saves nothing and does not change what you have marked read.
Finished reading?
0 of 36 chapters read in this series