Consensus and coordination
Some decisions must be made exactly once, by everyone, even while machines crash and messages get lost: who is the leader, which write came next, who holds the lock, which servers are in the cluster. That's the consensus problem. You'll rarely implement it, but nearly every system you design leans on it through etcd, ZooKeeper, a database's failover, or Kafka's controllers, and knowing how it behaves explains failover, split brain and why locks need fencing.
1. The idea
- Consensus: a group of nodes agrees on a single value (or a sequence of values, a replicated log), and once decided it never changes, even if some nodes crash or messages are delayed.
- Majority (quorum): with N nodes, decisions need ⌊N/2⌋ + 1. A cluster of 3 tolerates 1 failure; 5 tolerates 2; 7 tolerates 3. Even sizes add cost without adding tolerance (4 still tolerates only 1).
- Why it's hard: a node can't tell a crashed peer from a slow network. A famous result (FLP) shows no algorithm can guarantee agreement in bounded time in a fully asynchronous network; practical algorithms use timeouts and stay safe (never disagree) while sometimes pausing.
- The algorithms: Paxos (the classic), Raft (designed to be understandable; used by etcd, Consul, CockroachDB, TiKV and Kafka's KRaft) and ZAB (ZooKeeper's).
2. How it works
Raft in three ideas
- One leader per term. Time is divided into numbered terms. If followers stop hearing the leader's heartbeats for a randomised timeout (the Raft paper suggests 150–300 ms), one becomes a candidate and asks for votes; a majority makes it leader for the new term. Each node votes once per term, so a term has at most one leader.
- The leader appends, followers copy. Clients send commands to the leader, which appends them to its log and sends them to followers. An entry is committed once a majority has stored it; only then is it applied and acknowledged.
- Safety rules. A candidate only wins if its log is at least as up to date as the voters', so a new leader always has every committed entry. A stale leader that comes back sees a higher term and steps down.
Press Play to watch an election, replication, and a leader failure.
What you build on top of consensus
| Need | How a consensus store provides it |
|---|---|
| Leader election | Candidates try to create a key with a lease (etcd) or an ephemeral node (ZooKeeper); whoever succeeds is leader until the lease expires or the session dies. |
| Distributed lock | The same mechanism, for a resource: one holder at a time, released on expiry. |
| Configuration and feature flags | Small, strongly consistent values that services watch for changes. |
| Membership and service discovery | Each live instance keeps a lease-backed key; the set of keys is the set of healthy members. |
| Metadata for bigger systems | Which node owns which shard, which broker leads which partition (Kafka KRaft), the whole cluster state (Kubernetes stores it in etcd). |
Leases and fencing tokens
A lock or leadership lease expires after, say, 10 seconds unless renewed, so a crashed holder can't block everyone forever. But a holder that's merely paused (a long garbage-collection pause, a VM migration) may wake up after its lease expired and someone else took over, and still believe it holds the lock. Now two clients act as owner.
The fix is a fencing token: every time the lock is granted, the lock service hands out a number that only increases (etcd's revision, ZooKeeper's zxid). The holder sends it with every write, and the storage rejects any token lower than the highest it has seen. The paused client's late write carries an old token and is refused. Try it in section 3.
3. Try it yourself
How big should the cluster be?
A lease, a pause and a fencing token
Client 1 holds a 10-second lease and then freezes for a while. Client 2 gets the lease when it expires. Then client 1 wakes up and writes.
4. Trade-offs
Safety over liveness
Consensus never lets two nodes decide differently, but it can stall: without a majority, it stops accepting writes. That's the CP choice from Module 16, on purpose.
Every write waits for a majority
Latency is set by the round trip to the closest majority. Spread nodes across zones (about a millisecond apart), not across continents, unless you accept tens of milliseconds per write.
Small and important
Consensus stores are built for kilobytes of critical metadata and modest write rates, not for application data. Keep the big data elsewhere and store pointers and leases here.
Use a service, don't write one
Correct consensus code is famously hard. Use etcd, ZooKeeper, Consul or your platform's equivalent, or a database that runs consensus internally.
5. In real systems
- etcd (Raft) is Kubernetes' brain: every object in a cluster is stored there. Its FAQ recommends an odd number of members and no more than about seven, because each write must reach a majority.
- ZooKeeper (ZAB) coordinated Hadoop, HBase and, for over a decade, Kafka. Kafka 4.0 removed that dependency: its own Raft-based KRaft controllers now manage cluster metadata.
- Consul (Raft) provides service discovery, health checking and key-value configuration.
- Databases run consensus inside: CockroachDB, TiDB/TiKV and YugabyteDB run a Raft group per range of data; Spanner uses Paxos per split. Managed failover tools for PostgreSQL (Patroni) use etcd or Consul to elect the primary.
- Redis-based locks (
SET key value NX PX 30000) are fine as an efficiency optimisation ("don't do this job twice, usually"), but Redis isn't a consensus system; for correctness, use fencing tokens checked by the storage, or a consensus-backed lock.
The technologies up close
Leader election through etcd with a lease; the storage checks a fencing token; three or five nodes across zones.State the failure model you assume (crash-recovery and an unreliable network), and don't explain Raft unless it's an infrastructure interview.
Failure models
The vocabulary · say which one you assumeA failure model states how components are allowed to fail. Every distributed algorithm is correct only under the model it assumes.
How it works
| Model | What a node may do | Nodes needed to tolerate f faults |
|---|---|---|
| Crash-stop | Works correctly, then stops forever | f + 1 for replication; 2f + 1 for consensus (a majority must survive) |
| Crash-recovery | Stops, then comes back, possibly having lost what wasn't on disk | 2f + 1; durable logs decide what survives a restart |
| Omission | Drops some messages it should send or receive | Handled by retries, acknowledgements and timeouts |
| Timing (performance) | Answers, but too late: GC pauses, overload, a slow network | Timeouts, leases with fencing; you can't tell slow from dead |
| Byzantine | Anything at all, including lying or sending conflicting messages | 3f + 1 (blockchains, some aerospace systems) |
Real data-centre systems assume crash-recovery with timing failures over an unreliable network, and use checksums to turn corrupted data into detectable crashes. Byzantine tolerance is reserved for nodes that don't trust each other.
Status today
The standard taxonomy from distributed-systems research; Raft, Paxos and ZAB tolerate crash-recovery faults with 2f + 1 nodes, not Byzantine ones.
Use it when
- Explaining why a cluster has 3 or 5 nodes
- Infrastructure interviews: locks, key-value stores, consensus
Skip it when
- Product designs where "nodes can crash and the network can drop or delay messages" is all that needs saying
In the interview
I assume nodes can crash and recover and the network can drop or delay messages, so five nodes tolerate two failures; nobody is malicious, so no Byzantine tolerance.
- Proposing Byzantine fault tolerance for servers you run yourself: it triples the cost for a threat you don't have.
- Treating a timeout as proof a node is dead. It may only be slow; that's why leases need fencing tokens.
Failure detection: heartbeats and phi accrual
Situational · infrastructure designsHow nodes decide that another node has failed: by missing heartbeats, either against a fixed timeout or as a growing level of suspicion.
How it works
- Fixed timeout: no heartbeat for 10 s → dead. Too short and a GC pause causes a needless failover; too long and real failures linger.
- Phi accrual detector: each node learns the normal gaps between another node's heartbeats and outputs a suspicion level (phi) that rises the longer one is late compared with that history. Each application picks its own threshold: act early for cheap decisions, wait for high phi for expensive ones like failover.
- Randomised election timeouts (Raft's 150–300 ms) stop every follower from starting an election at once.
Status today
Phi accrual (Hayashibara et al., 2004) is used by Cassandra and Akka; Raft-based systems use heartbeats with randomised timeouts; Kubernetes uses liveness probes and node leases.
Use it when
- Cluster membership in databases, caches and coordination services
Skip it when
- Application-level designs, where the load balancer's health checks are enough
In the interview
Nodes gossip heartbeats and use a phi accrual detector, so a GC pause raises suspicion without triggering an immediate failover.
- One fixed, short timeout for an expensive action like failover: a slow node becomes a split-brain risk.
Coordination services: etcd, ZooKeeper, Consul
The default · don't build consensus yourselfSmall, strongly consistent stores built on consensus, used for leader election, locks with leases, configuration and service discovery.
How it works
Clients create a key with a lease (etcd) or an ephemeral node (ZooKeeper); it disappears when the client stops renewing, which hands leadership or a lock to someone else. Watches notify clients of changes. Every write goes through the leader to a majority, so they're built for small, important data, not high-volume traffic.
Status today
etcd runs Kubernetes; ZooKeeper coordinated Hadoop, HBase and (until Kafka 4.0) Kafka; Consul adds service discovery and health checks. Run 3 or 5 members across zones.
Use it when
- Leader election, distributed locks for correctness (with fencing), cluster membership, configuration
Skip it when
- Storing application data or anything high-volume
- Locks that only save duplicate work (a Redis lock is fine there)
In the interview
The scheduler elects a leader through an etcd lease; the database rejects writes with an older fencing token.
- Implementing Paxos or Raft yourself in a product interview.
- Putting high-volume data in ZooKeeper or etcd: every write needs a majority.
The Two Generals problem
The default explanation for "why can't we be sure?"Two parties that can only talk over a link that may lose messages can never be certain they agree. It's the simplest impossibility result in distributed systems, and the reason "exactly once" is always built from retries plus deduplication, never from a perfect acknowledgement.
How it works
- The story: two armies on either side of a valley win only if they attack together. Messengers crossing the valley may be captured. General A sends "attack at dawn". If it arrives, B must tell A it arrived, or A won't risk attacking. But B can't be sure its acknowledgement arrived, so B needs an acknowledgement of the acknowledgement, and so on forever. Whichever message is last, its sender can't know it got through.
- The proof idea: take the shortest protocol that works. The last message in it might be lost, and its sender must act the same either way, so the protocol works without that message too. That contradicts "shortest", so no finite protocol exists.
- Where you meet it: a client sends "charge £20" and the connection drops. Did the server charge? The client can't know. Every API that moves money or creates things faces this.
- What real systems do: accept uncertainty and make it harmless. Retry until acknowledged, and make the operation idempotent with a request ID, so a duplicate does nothing (Module 07, 24). Or keep a durable log and let both sides reconcile later (outbox, Module 22). TCP's handshake doesn't solve the problem either: it settles for "probably".
- Its relatives: Two Generals assumes honest parties and a lossy link. FLP (above) adds that with no timing bounds, even reliable messages plus one crash make guaranteed agreement impossible. Byzantine generals (failure models, above) adds liars.
Status today
First described by Akkoyunlu, Ekanadham and Huber (1975) and named by Jim Gray (1978). It isn't a product; it's the reason idempotency keys, retries and reconciliation jobs exist.
Use it when
- An interviewer asks for exactly-once delivery or a guaranteed acknowledgement
- Explaining why a payment API needs idempotency keys
Skip it when
- Don't recite the armies story at length: one sentence, then the practical fix
In the interview
Over a lossy network the sender can never know its last message arrived, that's Two Generals. So I don't try for exactly-once delivery: I retry until acknowledged and make the handler idempotent with a request ID, which gives exactly-once effect.
- Promising exactly-once delivery over a network.
- Adding another acknowledgement round to remove the doubt: it never does.
- Treating a timeout as "it failed": it means "unknown".
A lock service end to end: Chubby
Classic design · OSDI 2006Google's Chubby is the lock service the rest of this module builds towards: a small, highly available store of files and locks that other systems use to elect a leader and to publish who it is. It was built for coarse locks held for hours (who is the master), not for locking individual requests.
How it works
- A cell of five replicas running Paxos; one is the master and serves all reads and writes. Three of five must be up.
- Files are locks. Any small file or directory can be locked; the leader writes its own address into the file it locked, so finding the leader is just reading that file.
- Sessions and KeepAlives. A client keeps a session alive with periodic KeepAlive calls; if the session's lease runs out, its locks are released.
- Client caching with invalidation. Clients cache file contents; before a change, the master sends invalidations to everyone caching that file and waits, so caches are never silently stale.
- Sequencers and lock-delay. A holder can get a sequencer (lock name, mode and generation number) to attach to its requests, so servers can reject an old holder: the fencing token from section 2. For servers that don't check, a lock lost without a clean release can't be taken again until a lock-delay passes (clients choose it, up to a minute).
Status today
Not available outside Google, but its ideas are everywhere: Apache ZooKeeper and etcd play the same role for Kafka (before KRaft), Kubernetes and many databases. Facts checked October 2026 in the Chubby paper.
Use it when
- Leader election, "who owns shard 12", small configuration that many servers must agree on
- "Design a distributed lock manager"
Skip it when
- High-rate, short locks (per request or per row): use the database's own locking or a design without locks
In the interview
Leader election goes through etcd, as Chubby did for GFS: candidates race for one lock key with a lease, the winner writes its address there, and every request it sends carries the lock's revision so storage can fence an old leader.
- A lock service on the hot path of every request.
- Locks without a lease or a fencing token.
6. Remember it
How this hook works: what "Any two majorities share a member" means
What the slogan means
- "Any two majorities"
- With 5 nodes, any group of 3 is a majority. Pick any two such groups: they must overlap.
- "share a member"
- That shared node has seen both requests and will refuse to vote twice in a term or to accept a decision that contradicts one it already stored.
- (consequences)
- A minority side of a partition can't decide anything; a new leader always meets someone who knows the latest committed entry; even-sized clusters waste a node.
Why this picture fits: it's the committee with minutes: two rooms can't both pass contradictory motions if each needs three of the same five people.
The story → the concept
| Committee | Consensus |
|---|---|
| Five members, three votes to pass | N = 5, quorum = 3 |
| The chair for this session | The leader for this term |
| The minutes | The replicated log |
| A motion passed and minuted | A committed entry |
| Two members in another room | A minority partition: can't decide |
| Numbered resolutions | Fencing tokens: later decisions supersede earlier ones |
Watch it on an example
- A job scheduler must run each nightly job once. Three scheduler instances compete for a leader key in etcd with a 10-second lease.
- Instance A wins (token 41) and starts jobs. A network blip isolates A; its lease expires, and B wins (token 42).
- A, still running, tries to mark a job "started" in the database with token 41. The database has already seen 42 and rejects it.
- No job runs twice, and A, seeing the rejection, stops acting as leader.
The trap it saves you from
Believing a lock with a timeout guarantees one holder. Time-based expiry plus pauses means two holders; only an overlap-based decision (and fencing) stops them acting together.
It still works when…
You size quorums in leaderless databases (Module 14): W + R > N is the same overlap idea applied to reads and writes.
7. Common mistakes
8. Design exercise (25 minutes)
- Design leader election for a service where exactly one instance sends daily invoice emails. Which store, what lease length, what happens on a pause, and how duplicates are prevented.
- Size a cluster: a metadata service must survive the loss of a whole availability zone in a three-zone region. How many nodes, placed where?
- Trace an election on paper: 5 nodes, the leader in term 3 dies after replicating entry 7 to two followers. Which nodes can win term 4? Is entry 7 committed?
- Try the fencing demo with the check off, then on. Write one line on why lease expiry alone isn't enough.
- Write it down: quorum = ⌊N/2⌋ + 1, and the fencing token rule.
9. Self-check quiz
How many node failures can a 5-node Raft cluster tolerate and keep working?
When is a Raft log entry committed?
How does a fencing token stop a paused former lock holder?
Is a 4-node consensus cluster more fault-tolerant than a 3-node one?
How many nodes does a crash-tolerant consensus cluster need to survive 2 failures?
A node's reply arrives after 30 seconds. What can the sender conclude?
A payment request times out. What does the client know?
Why can't adding an acknowledgement of the acknowledgement fix the Two Generals problem?
In Chubby, how do clients find out who the elected master is?
10. Go deeper
| Resource | Time | Why |
|---|---|---|
| The Secret Lives of Data: Raft | 15 min | An animated, step-by-step walk through elections and replication. |
| In Search of an Understandable Consensus Algorithm (the Raft paper) | 1.5 h | Unusually readable for a consensus paper; sections 5.1–5.4 are the core. |
| How to do distributed locking (Martin Kleppmann) | 25 min | Why leases need fencing tokens, with the GC-pause example. |
| etcd FAQ and ZooKeeper recipes | 30 min | Cluster sizing, failure tolerance, and how to build locks and elections on each. |
11. Under the hood (optional)
How consensus, locks, membership and conflict-free merging really work. Not needed for most interviews; open it when you want to know how the products really do it. All the deep dives are listed on the Under the hood page.
Consensus, coordination and CRDTsA hands-on Raft simulator, Paxos and ZAB, cluster sizing, leases and fencing (animated), ZooKeeper lock queues (animated), SWIM gossip, a CRDT merge lab, 8 products comparedabout 40 min
Consensus lets a group of machines agree on one ordered log of decisions even when some crash or the network splits. It sits under Kubernetes (etcd), Kafka's metadata (KRaft), distributed SQL (a Raft group per range) and every serious lock service. Knowing how it behaves under partitions explains what these systems can and can't promise.
Raft, hands on
Five servers. Make one time out to start an election, send client writes, split the network, and watch terms, logs and the commit index. Writes count as committed only once a majority holds them.
Try: elect S1, write twice, partition, write again on S1 (it can't commit), time out S3 to elect a new leader in the majority, write there, then heal. S1 steps down and its uncommitted entries are replaced. Committed entries never are.
- Elections: a follower that hears no heartbeat within a randomised timeout (150–300 ms in the paper; etcd defaults to 1 s with 100 ms heartbeats) becomes a candidate, increments the term and asks for votes. Each server votes once per term, and only for a candidate whose log is at least as up to date as its own. That rule guarantees a new leader already has every committed entry.
- Replication: the leader appends to its log and sends entries with the index and term of the entry before them; a follower whose log doesn't match rejects them, and the leader backs up until the logs agree, then overwrites the follower's divergent tail.
- Commit: an entry is committed once stored on a majority, and the leader only counts replicas for entries from its own term (older ones commit indirectly). Committed entries are applied to the state machine.
- Reads: a leader may have been deposed without knowing; linearizable reads confirm leadership with a heartbeat round (ReadIndex) or rely on a time-based lease.
- Operations: logs are compacted with snapshots; membership changes add or remove one server at a time (or use joint consensus); etcd adds a "pre-vote" step so a partitioned node can't disrupt the cluster with ever-higher terms when it returns.
Raft, Paxos and ZAB
| Protocol | Shape | Where |
|---|---|---|
| Paxos (single-decree) | Agree on one value: prepare/promise with a ballot number, then accept/accepted by a majority | The theory everything builds on (Lamport) |
| Multi-Paxos | A stable leader skips the prepare phase for a sequence of values | Google Chubby and Spanner (per data split), many in-house systems |
| Raft | Multi-Paxos-like, designed for understandability: strong leader, explicit terms, log matching | etcd, Consul, CockroachDB, TiKV, KRaft, RabbitMQ quorum queues, NATS JetStream |
| ZAB | ZooKeeper's atomic broadcast: a leader orders updates; a new leader first syncs followers to its history | ZooKeeper |
Cluster size: a majority of N tolerates ⌊(N − 1) / 2⌋ failures: 3 nodes survive 1, 5 survive 2, 7 survive 3. Even sizes add cost but no tolerance (4 still survives only 1), and every extra member makes each write wait for more replies, which is why consensus groups are 3 or 5 and large systems run thousands of small groups (one per range) rather than one big one.
Leases, locks and fencing tokens
- A lease is a lock with an expiry, so a crashed holder doesn't block everyone forever. But a holder can pause (GC, VM migration) past its expiry and wake up still believing it holds the lock.
- Fencing tokens fix that: every grant comes with a number that only increases (etcd revisions, ZooKeeper zxids or sequence numbers), and the protected resource rejects any request with a lower number than it has already seen.
- Redis locks:
SET key token NX PX 30000is fine for efficiency (avoid duplicate work). The multi-master Redlock algorithm was criticised (Martin Kleppmann, 2016) because it relies on timing and offers no fencing token, so for correctness use a consensus store or a database constraint.
Coordination recipes
- Leader election: the same lock recipe; the lock holder is the leader, and everyone else watches.
- Service discovery and configuration: instances register ephemeral keys (ZooKeeper ephemeral nodes, etcd keys attached to a lease) that disappear when they die; clients watch for changes instead of polling.
- Don't store data there: coordination stores hold kilobytes of metadata and handle thousands of writes a second, not your application's records (etcd's default request size limit is 1.5 MiB and its database 2 GiB by default).
Membership by gossip
- SWIM (used by Consul's Serf, memberlist and many others): each node pings one random member per round; if no reply, it asks a few others to ping it indirectly before suspecting it; suspicions and joins spread by piggybacking on these messages. Load per node stays constant as the cluster grows, and news reaches everyone in O(log N) rounds.
- Cassandra gossips cluster state every second and uses a phi-accrual detector to decide when a node is down.
- Gossip is eventually consistent: fine for membership and load information, not for decisions that must be unique (use consensus for those).
CRDTs: merging without coordination
Conflict-free replicated data types are designed so replicas can change independently (offline, in different regions) and always merge to the same result, in any order. Change each replica, then merge:
| CRDT | State | Merge |
|---|---|---|
| G-counter | One count per replica | Element-wise max; value = sum |
| PN-counter | Two G-counters (increments, decrements) | Merge each; value = P − N |
| LWW-register | Value plus timestamp | Keep the later timestamp (concurrent writes lose one) |
| OR-set (observed-remove) | Each add gets a unique tag; a remove deletes the tags it has seen | Union of adds minus seen removes: a concurrent add survives a remove |
| Sequence CRDTs (RGA, YATA) | Each character has a unique ID and a position relative to its neighbours | Collaborative text: Yjs, Automerge |
OT vs CRDTs: operational transformation (Google Docs) rewrites concurrent operations against each other through a central server that orders them; CRDTs need no central order, which suits offline-first and peer-to-peer apps, at the cost of metadata per element (D19 builds an editor).
Product by product
| Product or scheme | How it works | The detail worth knowing |
|---|---|---|
| etcd | Raft-replicated key-value store with MVCC revisions, watches and leases | All Kubernetes cluster state; revisions double as fencing tokens |
| ZooKeeper | ZAB-replicated tree of znodes; ephemeral and sequential nodes | Locks and elections from "lowest sequence number wins" |
| Consul | Raft for the catalogue and key-value data; SWIM gossip for membership | Strong state plus cheap membership |
| Chubby, Spanner | Paxos lock service; Multi-Paxos per data split | Chubby inspired ZooKeeper |
| Raft inside databases | CockroachDB and TiKV (a group per range), Kafka KRaft, RabbitMQ quorum queues; MongoDB elections are Raft-like | Thousands of small groups, not one big one |
| Cassandra | Gossip with phi-accrual detection; lightweight transactions use Paxos per partition | Conditional writes cost several round trips |
| Redis locks | SET NX with expiry; Redlock across masters | No fencing token: use for efficiency, not correctness |
| CRDT libraries | Yjs, Automerge; Redis Enterprise active-active; Riak data types | Merge without a server; OT (Google Docs) needs one |
Say it in the interview
- "What happens to writes on the minority side of a partition?" The old leader can append but never commit them, because it can't reach a majority; once the partition heals it steps down and its uncommitted entries are overwritten. Clients saw those writes time out, not succeed.
- "Why 5 nodes, not 4?" Both tolerate fewer than half failing; 4 tolerates 1 like 3 does, while 5 tolerates 2.
- "Is a Redis lock safe for a payment?" Not on its own: a paused holder can outlive its lease. Use a fencing token checked by the database, or a unique constraint.
In Raft, which candidate will a server vote for?
How many failed members can a 5-member consensus group tolerate?
In an OR-set, replica A removes "milk" while replica B concurrently adds "milk". After merging?
Sources
- Raft paper (Ongaro and Ousterhout)
- The Raft site and visualisations
- etcd: API guarantees and revisions
- etcd: tuning (heartbeat and election timeouts)
- ZooKeeper recipes
- Consul architecture
- The Chubby lock service (Google)
- Paxos Made Simple (Lamport)
- How to do distributed locking (Kleppmann)
- Redis: distributed locks
- SWIM paper
- CRDT resources
- Yjs documentation
- Automerge