Consensus and leader election

Consensus is how a group of servers agrees on one ordered log of commands, so that together they act like one reliable machine even when some crash or the network splits. Leader election is its most common use: exactly one node may coordinate in each numbered term, and everyone can tell when that changes.

Picture it first: a society committee and its register

Your housing society keeps its register of maintenance payments in five copies, one per committee member. Three rules keep the copies identical:

• One writer. Only the secretary writes new entries; the others copy them in the same order.
• Three of five. An entry is final once three members have copied it, so it survives any two notebooks going missing.
• Numbered terms. If the secretary goes silent too long, a member stands for secretary in the next term. Each member votes once per term, and only for someone whose register is at least as complete as their own.

When the old secretary returns and writes under the old term, the others say they have moved on. He becomes an ordinary member, and any entry only he and one friend had copied is struck out: it never reached three people.

Map it back: members are servers, the register is the replicated log, the secretary is the leader, terms are Raft terms, “three of five” is the majority quorum, and “who has paid?” is the state machine.

What consensus solves

Consensus asks servers that may each propose a value to decide on exactly one. Agreement: no two decide differently. Validity: the decided value was actually proposed. Termination: every server that has not crashed eventually decides.

One value sounds small; a sequence of values is what makes it useful. Run consensus once per log slot and you get a replicated log, identical on every server. Feed it into a deterministic state machine (a key-value map, a lock table, a config tree) and every replica reaches the same state. This replicated state machine underpins etcd, ZooKeeper and the per-shard replication inside CockroachDB and Spanner.

Those properties come in two kinds:

• Safety: nothing bad ever happens. No two servers apply different commands at the same index; a committed entry is never lost, through any crash, delay or partition.
• Liveness: something good eventually happens, such as electing a leader or committing a write. It may pause while the network misbehaves.

A good consensus protocol never trades safety for liveness; when in doubt, it stalls. That is the CP behaviour from CAP and PACELC.

Fault models, majorities and FLP

Raft, Paxos and ZAB assume crash faults: a server may stop, restart from disk, run slowly or lose messages, but never lies. Tolerating f Byzantine servers, which can tell peers different stories, needs at least 3f + 1 of them and far heavier protocols; inside your own data centre you rarely pay for that.

To survive f crashes, a cluster needs 2f + 1 members, and every decision needs a majority, ⌊n/2⌋ + 1. Any two majorities of the same cluster share at least one member, so whatever one majority decided is known to someone in the next. That shared server is the trick behind every safety argument below.

Even sizes buy nothing. Going from 3 to 4 raises the quorum to 3 but still tolerates one failure, and a 2–2 split leaves neither side a majority. The etcd FAQ recommends odd sizes and no more than seven members; five, which Google’s Chubby also recommends, tolerates two failures.

FLP: safety always, progress only when the network behaves

In 1985 Fischer, Lynch and Paterson proved that in a fully asynchronous system, with unbounded message delays and no clocks, no deterministic protocol can guarantee that consensus terminates if even one process may crash. A crashed process looks exactly like a slow one, so an unlucky schedule of delays can keep the protocol undecided forever. FLP does not make consensus impractical; it says you cannot guarantee both safety and termination with no timing assumptions at all.

Real systems weaken the model, never the safety: timeouts suspect failures, randomisation breaks ties, and they assume partial synchrony, a network that eventually behaves long enough to finish an election. Timing affects only liveness; Paxos Made Simple notes that, because of FLP, electing a leader reliably needs randomness or real time.

Raft: terms and leader election

Raft (Ongaro and Ousterhout, 2014) was designed to be easier to understand than Paxos with the same guarantees. Each server is a follower (passive; redirects clients), a candidate (seeking votes) or the leader (handles every write and replicates it).

Time is divided into numbered terms, each starting with an election and having at most one leader. Terms work as a logical clock: every message carries the sender’s term, a server seeing a higher term adopts it (a leader or candidate also steps down), and a lower term is rejected.

How an election runs

1. The leader sends heartbeats: AppendEntries calls with no entries.
2. A follower that hears nothing for its election timeout increments its term, votes for itself and sends RequestVote to everyone.
3. Each server grants one vote per term, first come first served, written to disk before replying so a restart cannot make it vote twice.
4. A voter refuses if its own log is more up-to-date than the candidate’s: compare the last entries’ terms, and if equal, the longer log wins. This rule protects committed data.
5. A candidate with votes from a majority of the full cluster becomes leader. If it hears from a leader with a term at least its own, it reverts to follower.
6. If nobody wins, the vote was split; candidates time out again and start a new term.

Identical timeouts would split votes forever, so Raft draws each at random from a range; the paper’s example is 150–300 ms. Usually one server fires first and wins before others wake. On five servers, the paper measured elections consistently taking over 10 seconds with no randomness, a median of 287 ms with just 5 ms of randomness (150–155 ms), and 35 ms on average with 12–24 ms timeouts.

The timing rule is broadcastTime ≪ electionTimeout ≪ MTBF: the timeout must be long enough that a healthy leader is never suspected, yet short enough that the cluster is rarely leaderless. etcd defaults to a 100 ms heartbeat and a 1,000 ms election timeout, and its tuning guide wants the timeout to be at least ten round trips. Writes pause for about one election timeout after a leader crash. Detecting dead leaders is covered in heartbeats.

Raft: log replication and the commit rule

Only the leader accepts writes. For each client command it:

1. Appends the command to its log, tagged with the current term and next index.
2. Sends AppendEntries to every follower in parallel, with the new entries plus the index and term of the entry just before them (prevLogIndex, prevLogTerm).
3. Lets followers check. A follower accepts only if it holds an entry at prevLogIndex with term prevLogTerm. Otherwise the leader steps that follower’s nextIndex back and retries until the logs match; the follower then drops conflicting entries and appends the leader’s.
4. Tracks a matchIndex per follower: the highest index known to be stored there.
5. Commits. Once an entry is on a majority, leader included, the leader advances commitIndex, applies the entry and answers the client. Followers learn the new commitIndex from later AppendEntries calls.

The check works like induction: two logs sharing an entry with the same index and term are identical up to it (Log Matching).

One rule is easy to miss: a leader counts replicas only for entries from its current term. The Raft paper shows a crash sequence in which an older entry already on a majority is overwritten by a later leader. Once a current-term entry commits, everything before it is committed too, which is why a new leader immediately appends a blank no-op entry.

A write therefore costs one round of AppendEntries to the nearest majority, and a slow follower slows nobody down.

What happens to a deposed leader’s uncommitted entries

If a leader crashes before an entry reaches a majority, the next leader may overwrite it. The client just sees a timeout, and the write may or may not have happened, since the entry survives if a server holding it wins. A retry can therefore apply twice. The Raft paper’s fix: tag each command with a client ID and serial number, and have the state machine remember the last serial applied per client. See idempotency.

Why Raft is safe, even during a partition

Say entry e committed in term T, so a majority stored it. Any later leader won votes from a majority, and the two majorities share at least one server that stored e and then voted. It votes only for a log at least as up-to-date as its own, and the paper shows such a log must contain e. So every future leader holds every committed entry (Leader Completeness), and since servers apply entries in order, no two ever apply different commands at the same index.

The same overlap explains partitions. Take five servers with S1 leading term 3, and cut S1 and S2 off from the rest.

Raft does not promise that only one server believes it is leader: for a while S1 still thought it led term 3. The guarantee is Election Safety, at most one leader per term, plus the rule that only a leader in contact with a majority can commit. Inside the cluster the stale leader is harmless, because followers reject its lower term.

Split brain, and terms as fencing tokens

Outside the cluster, though, if the leader runs payouts or writes to object storage, a deposed leader that has not noticed yet, perhaps frozen in a GC pause, can still send those requests. The cure: attach the term to every side effect and have the receiver reject anything lower than the highest it has seen. A monotonically increasing term is exactly a fencing token; distributed locking covers the pattern in depth.

Paxos and ZAB in brief

Single-decree Paxos, set out in Lamport’s Paxos Made Simple (2001), chooses one value. Every proposal carries a unique, increasing proposal number; proposers draw from disjoint sets, such as a counter plus server ID.

1. Prepare and promise. A proposer sends prepare(n) to a majority of acceptors. Each acceptor that has not answered a higher number promises not to accept proposals below n, and reports the highest-numbered proposal it has accepted.
2. Accept and accepted. With a majority of promises, the proposer sends accept(n, v), where v must be the value of the highest-numbered proposal reported, or its own if none was. Acceptors accept unless they have since promised a higher number. A majority of acceptances chooses v.

That “must” is the overlap trick again: any later majority includes someone who accepted v, so the next proposer re-proposes it. Rival proposers can pre-empt each other forever with ever higher numbers (FLP in action), so Lamport adds a distinguished proposer: a leader.

Multi-Paxos runs one instance per log slot. A stable leader runs phase 1 once for all future slots, so each command costs only phase 2, one round trip to a majority, like Raft. Paxos leaves leader election, log gaps and membership open; Raft spells them out. Google’s Chubby and Spanner are Paxos-based. ZAB, ZooKeeper’s protocol, is a third leader-based design whose epochs do the job of terms: a would-be leader needs a majority before starting one.

Leader election for your own services

You rarely implement Raft just to pick a leader for a scheduler. You borrow a store that already runs it and use two features: a lease or session that vanishes when its holder stops renewing it, and ordered creation, so candidates queue fairly.

etcd: lease-bound keys and the election API

A client creates a lease with a time-to-live and keeps renewing it. Campaign writes a key under the election prefix, bound to that lease, and blocks until every older key is gone: the lowest creation revision leads, and each waiter watches only the key just ahead. If the leader dies, its lease expires, its key is deleted and the next candidate takes over; Observe streams each change to anyone watching. In Go:

sess, err := concurrency.NewSession(cli, concurrency.WithTTL(10)) // 10 s lease, renewed in the background
if err != nil {
	return err
}
defer sess.Close()

election := concurrency.NewElection(sess, "/elections/invoice-scheduler/")
if err := election.Campaign(ctx, hostname); err != nil { // blocks until this process leads
	return err
}
token := election.Rev() // our key's creation revision: higher for every new leader
go runScheduler(ctx, token)

<-sess.Done() // lease expired or revoked: we are no longer the leader
stopScheduler()

The creation revision doubles as a fencing token: etcd’s revision counter only grows, and each new leader’s key is younger than its predecessor’s.

ZooKeeper: ephemeral sequential znodes

The ZooKeeper recipe is the same idea. Each candidate creates a node under /election with the EPHEMERAL and SEQUENCE flags: ZooKeeper appends an increasing counter and deletes the node when its creator’s session ends. The lowest number leads; everyone else watches only the node just below their own, avoiding the herd effect of every candidate waking at once.

Kubernetes: Lease objects

Kubernetes control-plane components elect leaders through a Lease object stored in etcd; candidates race to write their identity into holderIdentity with a compare-and-swap, so one wins. kube-controller-manager defaults to a 15 s lease, 10 s renew deadline and 2 s retry period: the leader renews every 2 s and gives up after 10 s of failures, while others wait 15 s after the last renewal they saw. That 5-second margin is measured on each machine’s own clock, so it assumes clocks tick at similar rates.

Choosing an approach

Going deeper: cost, cluster size and geography

A consensus write cannot finish faster than one round trip from the leader to the nearest majority, plus a durable flush (fdatasync). etcd’s performance guide puts a data-centre round trip at a few hundred microseconds and an SSD flush often under 1 ms, so a three-member cluster answers a lightly loaded request in under a millisecond. A leader on a slow disk cannot commit in time, however promptly it heartbeats.

Throughput comes from batching, not from more nodes. One client writing sequentially at etcd’s measured 1.6 ms per write manages at most 1 ÷ 1.6 ms ≈ 625 writes/s (it measured 583); 1,000 concurrent clients, batched, reached 44,341 writes/s at 22 ms each. Extra members buy fault tolerance, not capacity: seven make every write wait for four acknowledgements, so etcd suggests stopping there.

Across regions, the majority sets the speed. Take five members, two in Mumbai, two in Hyderabad and one in Singapore, with illustrative round trips of 1 ms within Mumbai, 20 ms Mumbai–Hyderabad and about 60 ms from either city to Singapore. A Mumbai leader needs acknowledgements from Mumbai (1 ms) and Hyderabad (20 ms): about 20 ms plus the flush per commit. A Singapore leader’s two nearest followers are both about 60 ms away, so writes slow to about 60 ms, and ten round trips puts the election timeout at 600 ms or more. Keep the majority close together and steer leadership towards it.

Going deeper: reads, membership and disruptive servers

Linearizable reads

Reading from the leader’s memory is not enough: a deposed leader across a partition would serve stale data. Three safe options, slowest first:

• Through the log. Append the read as an entry and answer once it commits: a disk write per read.
• ReadIndex. The leader notes its commitIndex, confirms it is still leader with one heartbeat round to a majority, waits until that index is applied, then answers. No disk write, one round covers a batch, and followers can ask the leader for the index and serve reads themselves. In etcd’s benchmark one client saw 0.7 ms per linearizable read against 0.3 ms for a serializable read, answered locally by any member and possibly stale.
• Leader leases. After a majority acknowledges a heartbeat, the leader assumes nobody else can win for about one election timeout and answers locally. Fastest, but safety now rests on bounded clock drift, which a VM migration or long pause can break.

Changing membership

Switching member lists in one step is unsafe: servers learn the new list at different moments, so an old and a new majority could each elect a leader. The Raft paper uses joint consensus, an interim configuration needing majorities of both lists; Ongaro’s thesis showed a simpler route, changing one server at a time, since any majority of n members overlaps any majority of n ± 1. Add new members as non-voting learners (etcd’s term) until they catch up. To replace a dead member of a three-node cluster, remove it first: adding first gives four members, a quorum of three and only two healthy.

Pre-Vote and check-quorum

A server cut off from the others keeps timing out and raising its term; when it returns, that term forces a healthy leader to step down. With Pre-Vote, a would-be candidate first asks whether a majority would vote for it; servers that heard from a leader recently say no, so its term never rises. Check-quorum is the mirror image: a leader that has not heard from a majority within an election timeout steps down, so one stranded in a minority stops holding requests it can never commit.

When to reach for consensus

Use a consensus-backed store when being wrong means two of something that must be one:

• Leader election for singletons: a scheduler, a cron runner, a shard primary (design a job scheduler).
• Configuration and metadata: membership, partition ownership, settings that must flip everywhere at once. Kafka moved its metadata from ZooKeeper to a Raft-based quorum.
• Locks with fencing tokens (design a distributed lock).
• Unique ID ranges: hand out blocks of 1,000 IDs via compare-and-swap (design a unique ID generator).

Do not push ordinary application data through etcd or ZooKeeper. etcd is built for small data that fits in memory (a 2 GB default quota, 8 GB suggested maximum), and one cluster tops out in the tens of thousands of writes per second. Distributed databases do run consensus, but in many small groups: CockroachDB and TiKV run one Raft group per key range, and Spanner uses Paxos groups the same way. Data that tolerates merging can skip consensus with leaderless replication.

Cheat sheet

How to say it in the interview

“Consensus is how replicas agree on one ordered log despite crashes, so the same deterministic code on every node reaches the same state. I’d use Raft through etcd, not my own implementation. One leader per term replicates entries; an entry commits once a majority has it on disk. Five nodes survive two failures, and a write costs a round trip to the two nearest followers plus an fsync, a millisecond or two in one region. If the leader dies, followers wait a randomised timeout and a candidate with an up-to-date log and a majority wins. Majorities overlap, so committed data is never lost and a partitioned old leader can’t commit. Our scheduler campaigns on an etcd lease, and the winner passes its key revision as a fencing token.”

Why five nodes, and not four or six? Four nodes need three votes, so they tolerate one failure, like three, and a 2–2 split stops everything. Five survives one node in maintenance plus one crash.

Can a Raft cluster ever have two leaders? Briefly, in different terms: a deposed leader may not know yet. It cannot commit, because the followers it needs have moved to the higher term. Outside the cluster, fence its side effects with the term.

The leader crashed just after a client sent a write. What does the client see? A timeout. If the entry reached a majority it survives; if not, it may be overwritten. The client retries with the same request ID and the state machine deduplicates.

How do you serve linearizable reads without writing to the log? ReadIndex: note the commit index, confirm leadership with one heartbeat round, wait until that index is applied, then answer. Leases skip the round but rely on bounded clock drift.

Would you stretch one cluster across regions? Only for small, critical metadata where surviving a region loss is worth tens of milliseconds per write. Otherwise run a cluster per region and replicate asynchronously between them.

Where to go next

• Distributed locking: leases, fencing tokens, and why a Redis lock is not consensus.
• Heartbeats: failure detection and timeout arithmetic.
• ZooKeeper and etcd: data models, watches and operations.
• CAP theorem and PACELC: why consensus-backed systems are PC/EC.
• Data replication: leader-based replication and its failover risks.
• Clocks, ordering and conflict resolution: ordering without consensus.
• Distributed transactions: 2PC vs sagas: agreement across services, and why 2PC blocks.

References and further reading

1. In Search of an Understandable Consensus Algorithm, extended version (Ongaro and Ousterhout, 2014)raft.github.ioThe Raft paper: roles, terms, randomised 150–300 ms election timeouts, the vote and commit rules, the safety argument, joint consensus and the election-time measurements.
2. Consensus: Bridging Theory and Practice (Ongaro, PhD thesis, 2014)web.stanford.eduSingle-server membership changes, Pre-Vote, leaders stepping down without a majority, ReadIndex and lease-based reads.
3. Paxos Made Simple (Lamport, 2001)lamport.azurewebsites.netPrepare/promise and accept/accepted, proposal numbers, the distinguished proposer, and Multi-Paxos as a sequence of instances.
4. Impossibility of Distributed Consensus with One Faulty Process (Fischer, Lynch and Paterson, JACM 1985)groups.csail.mit.eduThe FLP result: no deterministic protocol guarantees termination in a fully asynchronous system with one crash.
5. etcd FAQetcd.ioWhy odd cluster sizes, the majority and failure-tolerance table, at most seven members, remove-before-add, disk latency and leader loss.
6. etcd tuning guideetcd.ioDefault 100 ms heartbeat and 1,000 ms election timeout, and sizing both from the round-trip time.
7. etcd performance guideetcd.ioA commit costs a round trip plus fdatasync; benchmark numbers for writes and for linearizable versus serializable reads.
8. etcd API reference: concurrency (Lock and Election services)etcd.ioCampaign, Observe, Proclaim and Resign, leadership tied to a lease, and the leader key’s creation revision.
9. ZooKeeper Recipes and Solutions: Leader Electionzookeeper.apache.orgEphemeral sequential znodes, watching the next-lowest node, and avoiding the herd effect.
10. kube-controller-manager command-line referencekubernetes.ioLease-based leader election and its 15 s lease duration, 10 s renew deadline and 2 s retry period defaults.