Loading...
Loading...
Leader election and managing state machine replication
The ring moved 8M of 40M keys with a versioned handoff, yet two clients holding two ring versions still disagree on the owner. Placement cannot settle that dispute, so three replicas must agree that a card charge committed before answering the user. If the leader crashes mid-replicate, followers must elect a successor without diverging or charging twice. Consensus, meaning a protocol where majorities vote disparate machines into one agreed sequence, turns a scattered log into a single totally ordered commit history. Picture a restaurant kitchen where only the expediter calls out the ticket order: if the expediter faints mid-shout, the cooks need one rule for picking the next caller without cooking two different tickets.
Client → Leader: append x=5 Leader → Followers: AppendEntries(term, prevIndex, entries=[x=5]) Followers: if prevIndex matches → append, reply ack Leader: majority ack → commit → reply to client → followers commit via next heartbeat
A term, meaning a monotonically increasing election number, labels every leader reign so stale leaders are recognizable. A quorum, meaning any majority whose any two instances overlap in at least one node, is what makes commitment stick: once a majority holds an entry, any future majority includes a witness to it.
The naive approach lets any node accept any write, which diverges the moment two nodes accept different values. Both Raft and Paxos fix this with majorities, but they organize the voting differently. Raft, a consensus protocol built around a single elected leader plus terms plus a replicated log, funnels all writes through the leader. Paxos, an older consensus family where proposers run prepare and accept phases against acceptors, can ballot any value without a stable leader, which is more general and harder to follow.
| Concern | Behavior |
|---|---|
| Leader crash | New election; higher term wins. Uncommitted tail entries may truncate, which is safe because clients were never told they committed. |
| Split brain | Majority partition commits; minority stalls, so no dual commit through the commit path. |
| Membership change | Joint consensus, meaning requiring majorities of old and new configs together, avoids two live majorities. |
Your payment service runs 5 replicas and the network splits 3 versus 2. Both sides want to commit a $40 refund. Quorum math, meaning the rule that any two majorities overlap, stops the double spend: with N nodes, majority equals floor(N/2) plus 1, so any two majorities share at least one node and the second committer necessarily sees the first commit.
N=5 → majority = floor(5/2)+1 = 3 Partition A (3 nodes): can reach 3 votes → may commit Partition B (2 nodes): max 2 votes → stalls, returns error, never commits Any two groups of 3 in 5 nodes share ≥1 node → second committer sees first commit N=3 → majority 2 → tolerates 1 failure N=5 → majority 3 → tolerates 2 failures N=7 → majority 4 → tolerates 3 (rarely worth the write latency)
The same overlap powers quorum stores: if writes reach W replicas and reads consult R replicas with W plus R exceeding N, reader and writer sets intersect and the reader can find the latest write. Raft bakes this in by requiring a majority before calling anything committed. Adding nodes buys fault tolerance one failure at a time while charging every write an extra round trip, which is why 7-node clusters are rare outside global control planes.
Every follower runs a countdown: hear from the leader and reset, hear nothing and campaign. If every node used an identical 200ms timeout, three followers would campaign together, split the vote 1-1-1, and repeat indefinitely. Randomizing each timeout across 150 to 300ms breaks the symmetry so one candidate usually pulls ahead on the second round.
Five nodes lose their leader. Three followers time out together and each votes for itself, so nobody reaches 3. Fresh random timeouts usually separate them next round. Tight ranges tie often; wide ranges fail over slowly. The 150–300ms spread used by etcd, a consistent key-value store backing cluster coordination, is the production compromise.
A partition heals and briefly two nodes think they lead, because the minority-side leader has not yet heard the higher term. Its writes cannot commit without a majority, so they sit uncommitted and truncate when it steps down. Reads are the genuine hazard: serving from the stale leader returns pre-partition values as if current.
| Failure | What the client sees | Safe handling |
|---|---|---|
| Stale leader serves reads | Returns pre-partition value as if current | Leader leases or read-index, meaning the leader confirms majority contact before answering linearizable reads |
| Uncommitted entry on old leader | Client heard accepted but never committed | Acknowledge only past the commit index; clients retry with idempotency keys |
| Membership change mid-term | Old and new majorities could each commit | Joint consensus: require majorities of both configs until the switch commits |
Nobody runs a shopping cart on Raft, because the leader bottleneck caps writes at thousands per second, not millions. Consensus guards small precious state such as leader-election records, shard maps, locks, and configuration. The data plane shards by key with consistent hashing plus quorum replication; membership gossips softly. Anchors that follow this split include etcd behind Kubernetes, both cluster-coordination systems, Consul locks, and CockroachDB ranges, meaning sharded SQL ranges each with local consensus. ZooKeeper runs Zab, a Raft cousin for coordination, and Spanner, Google's globally consistent database, runs Paxos underneath.
| Layer | Mechanism | Example |
|---|---|---|
| Control (small, precious) | Raft, 3–5 nodes, majority commit | etcd behind Kubernetes, Consul locks, shard-map masters |
| Data (large, sharded) | Consistent hashing plus quorum replication | Cassandra ranges, Dynamo partitions, CockroachDB per-shard Raft |
| Membership (soft state) | Gossip plus SWIM probes | Cassandra failure detector, Redis Cluster slot gossip |
A Raft log that never compacts fills any disk: ten thousand writes per second accumulate hundreds of millions of entries daily. Snapshotting compacts them by packaging the state machine as of some index into one blob that lagging followers install instead of replaying millions of AppendEntries. The log becomes a small tail plus a snapshot, not an infinite tape.
log: [1..5,000,000 applied] + tail [5,000,001..5,010,000] snapshot at 5M: state blob (e.g., 200MB kv image) + lastIncludedIndex=5M, term=41 follower 12M entries behind → leader sends InstallSnapshot RPC, not 12M AppendEntries follower installs, discards log < 5M, resumes from tail Knobs: snapshot every 10k–100k entries or 100MB–1GB of log (etcd: snapshot-count 100k) too rare → slow catch-up + disk growth; too often → snapshot CPU steals write tail
Linearizable reads, meaning reads guaranteed to reflect all completed writes, through the log would halve throughput. Production Raft uses read-index, where the leader confirms it still leads with one heartbeat round then serves locally, or lease reads inside its election lease. A deposed leader cannot reach a majority, so it cannot answer either way.
Moving from A,B,C to C,D,E enters a joint config requiring majorities of both, commits that, then switches to the new config alone. The overlap node C plus dual majorities forbid old and new clusters from each electing a leader at once.
A partitioned leader keeps heartbeating its two-node minority, convinced it still rules, while the majority elects a successor and commits writes it never sees. When the partition heals, the higher term forces it down on first contact. Two guards keep the interregnum safe. PreVote, meaning campaigning without bumping the term to test winnability, stops minority boxes from disrupting healthy leaders. CheckQuorum, meaning a leader stepping down when it stops hearing from a majority, shrinks the stale-read window to about one heartbeat round.
A timed-out node asks whether a majority would grant votes without incrementing its term. A partitioned minority gets no majority and stays follower with term untouched. Only a granted pre-vote starts a real election, so flapping network cards stop deposing leaders.
A leader that stops hearing from a majority steps down voluntarily instead of serving stale reads until out-termed. Combined with lease-bounded reads, discovery shrinks from indefinite to one heartbeat round. etcd, a consistent key-value store, ships both flags.
The leader appends a $40 refund at index 87 but crashes after one ack. Commit index, meaning the highest index stored on a majority, has not passed 87, so it stays uncommitted and the new leader may overwrite it. The client rule follows directly: acknowledge success only after the commit index passes the entry, and make every write idempotent, meaning safely retryable with the same key, so the retry lands exactly once either way.
Leader L1: log [..., 87:refund] on L1 + F1 only (2 of 5, NOT committed) L1 crashes → F2 elected (term+1, has entries ≤86 committed) → 87:refund overwritten client that heard "ok" from L1 pre-crash must retry with idempotency key → single refund Rule: commitIndex = highest index on a majority; ack clients only past commitIndex.
Committed entries, by contrast, survive every future election, because any new leader needed votes from a majority overlapping the commit majority and therefore contains them. That asymmetry is the whole contract: uncommitted may vanish, committed never does. Raft gives Paxos-grade safety with a leader plus terms plus one log rule, bought with a write bottleneck at the leader and a practical ceiling near 5 nodes before sharded consensus or gossip takes over.
Seven nodes tolerate three failures where five tolerate two, so the naive design keeps adding voters for safety. Rejected alternative: one 7-node group for the whole fleet. Every write then waits on 4 acks including the slowest voter and elections involve all 7, while the leader bottleneck caps throughput regardless of voter count. The working pattern shards consensus instead: CockroachDB-style ranges, meaning sharded SQL key ranges each with local Raft groups, run many small 3-to-5-node consensuses rather than one large one. Membership changes flow through joint consensus one shard at a time, so no global election ever stalls the fleet. Add groups, not voters, past five.
1 group × 7 nodes: majority 4, tolerates 3, every write waits 4 acks + slowest election 100 ranges × 3 nodes: each majority 2, tolerates 1 per range, writes parallel by range range failure: only its 3-node group re-elects (~300ms); other 99 ranges unaffected cost: 100 leaders to track vs 1; placement and shard maps become the new control plane. rule: quorum size guards one shard; sharding guards the fleet.
Raft terms order one group log, and 100 ranges carry 100 independent terms that never compare. When a withdrawal in one range and a deposit in another hit 150ms apart, which term decides the real order across groups?