Consensus, Leader Election and Coordination

It is 10:00:04 on a tatkal morning, and a railway booking site is taking 20,000 booking requests a second. The primary database that holds coach B4's seats freezes for eight seconds in a garbage-collection pause. Its standby hears no heartbeat for five seconds and promotes itself. Then the old primary wakes up, unaware that any time has passed, and keeps selling. For the next half minute two primaries accept bookings, and seat 23 in coach B4 is sold to Asha on one and to Vikram on the other. Two passengers will board with tickets for one berth, and someone will untangle it by hand.

This chapter follows that booking site. Many of its parts work only when exactly one machine is in charge, and each breaks in its own way when two act at once:

Every row needs the same guarantee: at most one owner at a time, even while machines crash, freeze and lose touch.

The obvious design is one coordinator that decides for everyone: simple, fast, and a single point of failure. The real question is how to add a spare without ever creating two owners.

The natural fix is a spare. A primary sends a heartbeat every second, and a standby promotes itself after 5 seconds of silence. It is like a night watchman who decides his partner has gone home because the phone line went quiet. That works when the primary crashes, and fails when it is only slow or cut off.

This is split brain: two nodes each believe they lead and accept writes the other never sees. It needs no network fault at all; the tatkal outage was a pause, and to a paused process no time has passed.

Tuning the timer does not save it. A shorter timeout promotes the standby during ordinary pauses; a longer one leaves the site without writes for longer after a real crash. As the abstractions chapter put it, a timeout cannot tell a slow node from a dead one.

The cure has two halves: let a majority of a small group decide who leads, so two sides of a split cannot both win, and number every leadership, so an old leader that wakes up late can be refused. The next part starts with majorities.

A majority works like a committee that needs more than half its members to pass anything: a room split in two cannot have more than half on both sides. Run a small group of nodes and let a decision stand only when a majority, also called a quorum, agrees. Any two majorities of the same group share at least one node, and that shared node will not agree to two conflicting things.

How many nodes? Each size survives a different number of failures:

Survival steps up only at odd sizes. This is the 2f + 1 rule: 2f + 1 nodes survive f failures.

An even count buys nothing: 4 nodes need 3 for a majority, so they survive one failure, like 3, while adding one more machine that can fail. Five survive two, which matters because one node is often down on purpose for an upgrade, and you still want to survive a crash during it.

Bigger groups cost more: a group of 7 waits for 3 followers on every write instead of 2. Most groups run 3 or 5 nodes.

A majority can vote, but on what? Consensus is getting a group of nodes to agree on values and never change their minds, even while some crash. Picture three clerks keeping copies of one register: if each writes the same lines in the same order, every copy reads the same.

In practice the nodes agree on a log, an ordered list of commands such as "sell B4-23 to Asha". Each node applies the log in order to its own copy of the data, so all copies end in the same state. This is called state machine replication.

The best-known protocol is Raft, built around one leader; the older Paxos family solves the same problem, and its common form also uses a leader. In Raft each node is a follower, a candidate or the leader. Time is cut into numbered terms, like numbered editions of a newspaper. Each term starts with an election and has at most one leader. Every message carries the sender's term, and two rules follow:

• A node that sees a higher term adopts it and becomes a follower at once, even if it was leading.
• A request stamped with a lower term is rejected, and the reply names the current term.

These rules are the numbering half of the cure. An old leader is refused by its term number, without anyone telling it that it was replaced. The next page shows how a term gets its leader.

Terms need leaders, and leaders are chosen by vote. The leader sends a heartbeat to every follower, say every 100 ms. Each follower runs an election timeout, picked at random from a range such as 1 to 2 seconds and reset by every heartbeat. When the timer runs out, the follower assumes the leader is gone and calls an election.

Two rules make the result safe. One vote per term means no term has two leaders: two winners would need two majorities, which share a node that voted once. Refusing a candidate with an older log means the winner holds every committed entry: that entry sits on a majority, the winner needed a majority of votes, and the two share a node that would have refused it.

The timeout sets the price of a failover: writes pause for about one timeout plus a vote, a second or two. Set it below a healthy leader's worst pause, such as a 2-second garbage collection, and followers call needless elections; set it at 10 seconds and every crash costs 10 seconds of writes. Pick the timeout above the worst normal pause, and accept the failover time that follows.

Once elected, how does a leader make a write stick? The leader handles every write. Asha's booking arrives, and the leader appends "sell B4-23 to Asha" to its log as entry 57 in term 4, then sends it to the followers. Each follower checks that its log matches the leader's up to entry 56, saves the entry to disk and replies.

Once a majority, counting the leader, has the entry on disk, it is committed: the leader applies it to the seat table, tells Asha her seat is booked, and tells the followers to apply it in its next message. A committed entry is never lost or changed, because every future leader needs a majority of votes, and that majority includes a node holding the entry.

The latency follows from the majority rule. With 3 nodes a write waits for the faster of 2 followers; with 5, for the second-fastest of 4. In one region that is a network round trip plus a disk flush, a few milliseconds, and one slow follower changes nothing.

Not every write commits. The leader repairs followers by resending from the last entry they agree on, and it overwrites entries an old leader never committed. So a write that timed out may or may not have committed, and the client cannot tell which; retries need the idempotency keys from the abstractions chapter. The next page puts these pieces together in a failover.

Here is the tatkal outage again, on a three-node consensus group. Node A leads term 4, commits Asha's booking, then freezes. B and C must carry on, and A must sell nothing when it wakes.

Nobody told A it had been replaced. The higher term did the work: C refused an append stamped with term 4, A stepped down, and B's log will overwrite A's uncommitted entry. The booking server got an error instead of a false "booked", and seat B4-24 has one owner. The double sale from the opening cannot happen, because a stale leader cannot gather a majority.

A frozen leader is one failure; a cut network is the other. What each side of a split can do depends only on whether it holds a majority. Take a 5-node group spread 2 + 2 + 1 over three zones:

In every case at most one side commits, and a side without a majority stops instead of guessing.

Reads need the same care. A cut-off leader does not yet know it was replaced, so answering from its own copy could show B4-24 as free after B sold it. It either confirms with a majority before answering, or holds a leader lease, a promise that no other leader can exist until a time has passed, which saves that round trip but trusts the clocks. Careful implementations also make a leader step down once it stops hearing from a majority, and use a pre-vote so a node returning from a split cannot force a needless election.

This is the abstractions chapter's trade during a split, consistency over availability: bookers on the minority side get errors, and no seat is sold twice.

Machines outside the group need roles too, such as which scheduler runs tonight's chart job. A holder that dies cannot hand its role back, so roles come as leases. A lease works like a library book with a due date: if you vanish, the book returns to the shelf on its own.

The grantor gives the holder a right for a fixed time, such as 10 seconds, and the holder renews it every 3 seconds or so. A dead holder stops renewing; once the lease expires, someone else may take the role, with no crash detection needed.

Lease length is a trade. With a 10-second lease, a crashed chart job waits up to 10 seconds for a standby. A 1-second lease fails over faster, but any hiccup longer than a second hands the role away, and renewals cost ten times the traffic.

Every lease is only as safe as the clock that times it. Three clock problems matter, each with a standard defence:

Healthy clocks differ in rate by well under 0.1%, a few milliseconds over a 10-second lease, so stopping a second early covers drift easily. No margin helps a holder that has stopped running, which is the next page's problem.

A process pause stops a program while the world moves on: garbage collection, usually for milliseconds and occasionally for seconds; a host pausing a virtual machine to move it; memory swapped to disk. The chart job can check its lease, freeze for 15 seconds, then write berths after its lease has passed to a standby.

The fix is a fencing token, which works like token numbers at a bank counter: once token 34 is being served, someone waving token 33 is turned away. The lock service issues a number with every grant, always higher than the last, and the holder sends it with every write. The storage rejects any token lower than the highest it has seen.

Worker 1 believed it held the lock, and from its own view it did. The berths table, which sees both writes in order, settled it: a token lower than the highest seen is refused, however sure its sender is. Next: where that check lives, and when to skip the lock.

Why must the storage check the token, and not the worker? Because a pause can land between any check the worker makes and its write; only the storage sees writes in order. In a database the check is one more condition on the update:

Often the lock itself is unnecessary. When the contested data lives in one database, a conditional write (compare-and-set) settles the race in one round trip: the database applies the write only if its stated condition still holds. Two passengers racing for B4-23 need no lock: UPDATE seats SET status = 'held' WHERE seat = 'B4-23' AND status = 'available' changes 1 row for one and 0 for the other. Choose by what a double run would do:

A lock for efficiency can be a plain lease; a lock for correctness needs a token the resource checks, or it protects nothing.

Who issues those leases and tokens? You rarely build consensus yourself. You run a coordination service: a small key-value store replicated by a 3- or 5-node consensus group, so every client sees changes in one order. Think of it as the notice board in a station master's office: short, official facts that everyone trusts. It holds megabytes to a few gigabytes, with values usually capped near a megabyte.

Five features do almost all the work:

Together they let a client claim a role, lose it automatically when it dies, and let everyone else notice within about a second.

Two details separate safe use from subtle bugs. First, a linearizable read goes through a leader that confirms it still leads, so it is never stale; a cheaper read from any node may lag. Use the first for "am I still the leader?".

Second, a watch says something changed, not what is true now. Notifications arrive late and can be missed during a disconnect. etcd-style watches resume from the last revision seen; older ZooKeeper-style watches fire once and must be set again. Either way, act through a conditional write, never on a notification alone.

With those five features, most coordination jobs follow one recipe: write a small fact conditionally, tie it to a lease if it should vanish with its owner, and let others watch it. The booking site uses it four ways:

Every row is small, changes rarely and matters to every server: that is what belongs here.

Locks with many waiters need one more trick. If 500 workers all wake when a lock frees, they stampede the service. Instead each creates a numbered entry and watches only the one just ahead, so a release wakes exactly one waiter, in arrival order.

Notice what is missing: bookings, passengers and payments. The service holds control facts, such as who leads and where data lives, never the user data itself.

Agreement is not free, so where should it sit? Each write costs a majority round trip plus a disk flush: a few milliseconds within a region. With batching, one group commits thousands to tens of thousands of small writes a second, plenty for control facts and far too few for user traffic.

Placement sets what the group survives and what each write costs:

Surviving a whole region costs every write a cross-region round trip, so keep a group per region and go global only for rare decisions. Keep the service off the hot path too: routers cache the partition map and watch it, and never call the service per booking.

A lock per request is the classic mistake. At 10 am the booking site runs 200 servers at 1,000 requests a second; a lock per request means 200,000 acquires a second plus the releases, ten to a hundred times what one group commits. Use consensus to decide who does the work, not to do the work.

Disks matter too: every commit waits for flushes, so a node sharing its disk with a busy logging agent slows every grant and can delay heartbeats into needless elections.

Consensus buys one agreed answer at the price of latency, three to five machines, and writes that stop when a majority is gone. Many problems do not need to pay that price. These are the common cases, with what to use instead:

Reach for consensus when a wrong answer is worse than no answer; otherwise choose something that stays available.

The rule cuts both ways: if two owners would double-sell seats or corrupt a partition map, a cheaper scheme that is usually right is the wrong choice.

Later chapters lean on both sides of this rule: the key-value store uses gossip for membership because brief disagreement is fine there, while unique IDs lease machine numbers, the task scheduler fences workers with attempt numbers, and group chat gives each conversation one sequence owner under a lease.

For "design a distributed lock service", spend the 40 minutes of design like this:

The depth is in pauses and failover: pick fencing for "what if a holder freezes?", failover for "what if the leader dies?", and queueing for 500 clients on one lock.