Consensus & coordination
Lesson 07 ended on a cliffhanger: linearizability requires the nodes to coordinate. This lesson is the machinery — how you force a set of machines, some of which crash or get partitioned, to agree on a single value that nobody can later contradict.
The problem, stated precisely
Strip away the products — leader election, distributed locks, a commit decision, the next ID in a sequence, "who owns shard 7" — and they are all the same primitive. You have N machines connected by an unreliable network. They must agree on one value. The contract:
- Safety (agreement): no two nodes ever decide different values. This must hold always, even during partitions, even if messages are delayed for hours. Violating safety means split-brain: two leaders, two owners of a lock, a transaction that committed on one side and aborted on the other.
- Liveness (termination): the cluster eventually decides, given enough non-faulty nodes and a network that eventually delivers messages. Liveness is the one we are allowed to compromise temporarily.
- Validity: the value decided was actually proposed by some node (no inventing answers).
The asymmetry matters: safety is unconditional, liveness is best-effort. A correct consensus system will rather stall (refuse to decide) than risk two answers. That single design choice is why a partitioned minority goes read-only instead of electing its own leader.
If the formalism feels abstract, hold this picture: consensus is like a board of directors trying to agree on one decision over unreliable phone lines. A decision becomes official only once a majority has signed off (safety); once it is signed, it cannot be quietly reversed later (durability of the decision); and nobody trusts a decision made by a lone director who couldn't reach the others (no minority acting alone). Keep that board in mind — every rule below is just that meeting, made precise.
Quorums: why a majority is the whole trick
The entire safety argument reduces to one geometric fact. A decision is committed only when a majority — ⌊N/2⌋ + 1 nodes — has accepted it. Any two majorities of the same set must share at least one node, because two disjoint subsets each larger than half would together exceed N. That shared node remembers the first decision and rejects any conflicting second one. Overlap is the safety mechanism; everything else is plumbing.
f = ⌊(N − 1) / 2⌋ failures tolerated · majority Q = ⌊N/2⌋ + 1 = N − f
This is why production clusters are odd-sized. An even N is strictly worse than the odd size below it: it needs a larger majority, tolerates the same number of failures, and gives the network more ways to split into two equal halves where neither side has a majority (total deadlock).
| N | Majority Q = ⌊N/2⌋+1 | Failures tolerated f | Verdict |
|---|---|---|---|
| 1 | 1 | 0 | No fault tolerance — one crash = total outage |
| 2 | 2 | 0 | Worse than 1: any single failure halts it |
| 3 | 2 | 1 | Good — the common minimum |
| 4 | 3 | 1 | Wastes a node — same f as N=3, larger quorum |
| 5 | 3 | 2 | Good — typical for important clusters |
| 6 | 4 | 2 | Wastes a node — same f as N=5 |
| 7 | 4 | 3 | Good — high tolerance, slower quorum |
Read it as a rule: going from one odd size to the next (3→5→7) buys exactly one more tolerated failure but enlarges the quorum you must round-trip on every decision. That is the durability-vs-latency trade you keep paying.
Raft — consensus you can actually reason about
Paxos (Lamport, 1989) is the original and provably correct, but notoriously hard to follow and to implement faithfully. Raft (2014) was designed for understandability and is what you should explain in an interview. ZAB (the ZooKeeper Atomic Broadcast protocol) is a close sibling. All three share the same safety core: majority overlap. They differ mainly in presentation.
Raft factors the problem into leader election + log replication, separated by a logical clock called the term (a monotonically increasing integer, one leader per term).
Leader election
- Every node is a follower, candidate, or leader. Followers expect periodic heartbeats from the leader.
- Each follower has a randomized election timeout (e.g. 150–300 ms). If no heartbeat arrives before it fires, the follower bumps the term, becomes a candidate, votes for itself, and requests votes from everyone.
- A node grants at most one vote per term. A candidate that collects a majority becomes leader and starts heartbeating.
- Randomization is the liveness trick: if two nodes time out together and split the vote, each re-randomizes its next timeout, so one will almost certainly go first next round. Split votes self-heal in a round or two.
Log replication
- Clients send writes only to the leader (followers redirect). The leader appends the command to its log and sends it to followers.
- An entry is committed once a majority has durably stored it; only then does the leader apply it to its state machine and answer the client. A committed entry is permanent — majority overlap guarantees any future leader already has it.
- If the leader dies mid-flight, the new leader's election rules ensure it holds every committed entry, so nothing committed is ever lost.
- The old leader dies. Its followers stop receiving heartbeats and start counting down their election timeouts.
- Node B's randomised timeout fires first. B increments the term to 5, votes for itself, and requests votes from the other four.
- B collects a majority — 3 of 5 votes — so B is now the leader for term 5 and begins heartbeating.
- A client's
set x=9arrives at B. B appends it to its own log as entry #8, still uncommitted — it has not yet touched its state machine and has told the client nothing. - B sends
AppendEntries(#8)to the followers. - Two followers store #8 durably and ack. With B itself that is 3 of 5 — a majority — so entry #8 is now committed.
- Only now does B apply
x=9to its state machine and reply "OK" to the client. The client never saw the value until it was safe. - Any future leader is guaranteed to already hold #8: to win an election its log had to be at least as up to date as the majority that stored #8, so a winner cannot be missing it. The "OK" is therefore permanent.
That is the safety rule from the start of the lesson made concrete: a value is decided only once a majority holds it, and majority overlap means no later leader can contradict it.
N = 5, leader-to-follower one-way network latency 1 ms, fsync 2 ms. A write commits when the leader plus Q − 1 = 2 followers have it durably.
- Leader appends + fsync: 2 ms.
- In parallel, leader ships to all 4 followers. It needs only the 2 fastest acks (the rest don't gate the commit). Each ack = 1 ms out + 2 ms follower fsync + 1 ms back ≈ 4 ms.
- Commit latency ≈ max(2, 4) ≈ 4 ms — one round-trip to the fastest majority, not to everyone.
The design rule: keep consensus off the hot path
Because each decision costs a majority round-trip and funnels through a single leader, consensus is the wrong tool for high-throughput data. The pattern that actually ships:
- Run a small, dedicated coordination service — ZooKeeper, etcd, or Consul — that uses consensus to agree on a tiny amount of critical metadata: who is the current leader, the cluster config, the shard-to-node map, lock ownership, service registry. In practice you build on a handful of primitives these services expose: watches (get notified the instant a key changes), leases / sessions (a key that auto-expires unless you keep renewing it — exactly the lock-lease above), compare-and-swap (atomically write only if the value is still what you last read — how you do leader election without a race), and ephemeral nodes (a key that vanishes the moment your session drops, so liveness is detected for you).
- The bulk data plane (your databases, your stream processors, your caches) reads that metadata cheaply — often cached locally with a watch — and never runs consensus on individual requests. Consensus decides policy; the data plane executes it at full speed.
| Job | Run consensus on it? | Why |
|---|---|---|
| Elect the primary of a DB shard | Yes | Rare, one decision, must be unique |
| Hold a distributed lock / lease | Yes | Safety-critical, low frequency |
| Store the shard map / config | Yes | Tiny, read-heavy, must be agreed |
| Every user write to the DB | No | Throughput-bound; replicate, don't vote per row beyond the shard's own group |
| Per-request rate-limit counter | No | Hot path; use a local/sharded counter (lesson 15) |
Locks, leases, and fencing tokens
A distributed lock from etcd/ZooKeeper is a lease: you hold it only while you keep renewing it. The danger: a client acquires the lock, then stalls — a GC pause, a VM freeze, a network blip — long enough for its lease to expire. The service hands the lock to a second client. Now the first client wakes up, believing it still holds the lock, and writes. Two writers, corruption.
The fix is a fencing token: every lock grant carries a monotonically increasing number. The protected resource (the storage layer) records the highest token it has seen and rejects any write carrying a lower token. The stalled client comes back with token 33; the resource has already accepted token 34; the stale write is fenced out. This is how consensus prevents the split-brain we worried about in lessons 08–09: the partitioned minority can't reach a majority, so it can't commit and can't get a fresh token, and the fencing check blocks whatever stale token it still holds.
- Holder #1 acquires the lock; the service hands it token 33 (a monotonically increasing counter).
- #1 hits a long GC pause (or VM freeze). It is frozen but still believes it holds the lock.
- The lease expires while #1 is paused. Holder #2 acquires the lock and is handed the next token, 34.
- #1 wakes up, oblivious, and writes to the shared storage stamped with token 33.
- The storage has already accepted token 34, so 33 < 34 and the write is REJECTED. The stale writer is fenced out — no second writer, no corruption.
That single monotonic check is what keeps a distributed lock safe across a pause: correctness no longer depends on #1 noticing in time that its lease lapsed.
Cluster sizing & cost calculator
Pick a cluster size and read off the safety and cost it implies. The math here is exactly the math you'd defend at a whiteboard.
Interview prompts you should be ready for
- Why are consensus clusters always an odd number of nodes? (senior answer) Majority Q=⌊N/2⌋+1; two majorities must overlap, which is what prevents conflicting decisions. An even N needs a bigger quorum yet tolerates the same f=⌊(N−1)/2⌋ as the odd size below it, and an even split deadlocks both halves. N=4 buys nothing over N=3 and costs more — strictly dominated.
- What does FLP impossibility mean for a real system? (senior answer) In a truly asynchronous network with one crash you can't guarantee termination, because you can't tell "slow" from "dead." Real systems add timeout-based failure detectors to regain liveness, accepting that a slow-but-healthy node may be wrongly suspected and trigger an election. Safety never depends on the timeout; only liveness/leadership stability does.
- Walk me through a Raft leader election. (senior answer) Followers run randomized election timeouts; on expiry a node bumps the term, becomes candidate, votes for itself, requests votes. One vote per node per term; a majority wins. Randomized timeouts make split votes rare and self-healing. The new leader is guaranteed to hold all committed entries by the election restriction.
- How does a write commit, and what's the latency floor? (senior answer) Leader appends, replicates, and commits once a majority has it durably; cost ≈ one round-trip to the Q-th fastest node plus fsync. It does NOT wait for all nodes. Throughput is capped by the single leader — consensus doesn't scale writes with N; more nodes buy tolerance, not speed.
- You need a distributed lock. How do you make it safe against a stalled holder? (senior answer) Lease-based lock plus a monotonic fencing token on every grant. The resource tracks the max token seen and rejects lower ones, so a client that stalled past its lease expiry and woke up is fenced out. A lock without fencing is a bug waiting for a GC pause.
- How does consensus prevent split-brain during a partition? (senior answer) The minority side can't assemble a majority, so it can't elect a leader or commit; it goes read-only. The majority side keeps working. Stale writes from a deposed leader carry old fencing tokens and are rejected. Two leaders can momentarily exist; two committed decisions cannot. That's choosing C over A in CAP.
- Why not just run everything through ZooKeeper/etcd? (senior answer) Every operation would pay a majority round-trip and serialize through one leader — fine for kilobytes of metadata, fatal for a data plane doing 100k writes/s. Use consensus for leader identity, config, locks, shard maps; let the data plane read that metadata cheaply and run at full throughput.
- When would you pick N=5 over N=3? (senior answer) When you must tolerate 2 simultaneous failures — e.g. an AZ outage plus a node crash during maintenance — and can absorb the larger quorum (Q=3) on the commit path. N=5 also tolerates one straggler without latency impact since you only wait for 3 of 5. The cost is more hardware and a slightly higher latency floor.