Step 2 of 14 · 1 min

Why "exactly one" is hard

On one machine, ensuring only one thread does something is easy: use a lock, which the operating system enforces. Across machines, there's no shared operating system, only messages over a network, and the network and the machines can fail in awkward ways:

• Messages can be delayed for an unknown time, or lost.
• A machine can freeze (a pause, a suspended VM, an overloaded CPU) and resume later as if nothing happened.
• Clocks drift and jump (lesson 4.3), so "my lease is valid until 10:03:15" depends on whose clock says it's 10:03:15.
• You can't tell a slow machine from a dead one. No reply could mean either.

There's a famous result, from Fischer, Lynch and Paterson in 1985, showing that in a fully asynchronous system (where messages can take arbitrarily long), no algorithm can always reach agreement if even one machine might fail. Real systems work around this by assuming messages usually arrive within some time, and making sure that when that assumption breaks, they become slow or unavailable, but never wrong.

Step 3 of 14 · 1 min

Leases and their weakness

A lease is a lock with an expiry time: "you hold this until 10:03:15, unless you renew it." Leases solve one problem well: if the holder dies, the lock is released automatically, so the system doesn't wait forever. That's why lesson 2.1's digest job used SET ... NX PX 600000 and lesson 4.3's generators used lease rows.

But they rely on the holder noticing that its lease has run out, and a frozen process can't notice anything. When it resumes, it carries on with whatever it was doing, which may be writing to shared data as if it still held the lock. Checking the lease just before acting doesn't fully help either, because the pause can happen between the check and the action.

Step 4 of 14 · 1 min

Fencing tokens

The fix is to stop relying on the holder to behave, and have the thing being protected check instead.

Every time a lease is granted, the lock service hands out a fencing token: a number that increases with each grant. Server api-42 got token 33 when it took machine ID 17. When api-77 took it over, it got token 34.

Then, whatever the lock protects must check the token on every write, and reject anything older than the newest token it has seen:

• api-77 writes with token 34. Accepted; the storage now knows 34 is the latest.
• api-42 wakes up and writes with token 33. Rejected: 33 is older than 34.

The holder doesn't need to know it was frozen. The storage refuses its stale writes. This idea, written up clearly by Martin Kleppmann in a 2016 post about distributed locking, is the missing piece in many lock designs. Lesson 5.4's "fencing the old leader" is the same idea applied to database leaders: each leader has a term number, and storage or replicas reject writes from leaders with an older term.

For Chirp's ID generators, there was no single storage to check tokens against: IDs go into many tables. So the team changed the design. Each generator includes its lease's token in a heartbeat it must confirm before issuing each batch of IDs, and stops for good (exits and restarts) if a confirmation ever fails or is late. It also refuses to generate if more than a few seconds have passed since its last confirmed heartbeat, by its own monotonic clock, which keeps counting through a pause. The window for duplicates shrank from "as long as a pause" to almost nothing. The primary keys remained the last line of defence.

Step 5 of 14 · 3 min

Agreeing on a leader, properly

Leases and fencing tokens assume something hands out the leases and the increasing tokens, and that it doesn't have two leaders. Something at the bottom has to be genuinely reliable. That's a consensus system: a small group of machines that agree on decisions, such as who the leader is, what the next token is, or what comes next in a log, in a way that stays correct through crashes, freezes and network splits.

Lesson 5.4 gave the key ingredient: majorities. With 5 machines, any decision needs 3 to agree. Two different majorities of 5 always share at least one machine, so two conflicting decisions can never both be made.

The two best-known consensus algorithms are Paxos, described by Leslie Lamport (first in 1989, with the well-known "Paxos Made Simple" in 2001), and Raft, designed in 2014 by Diego Ongaro and John Ousterhout specifically to be easier to understand. They give the same guarantees. Here's how Raft works, in plain terms.

Raft: terms and elections

A Raft group has (usually) 3 or 5 servers. At any time, each is a follower, a candidate, or the leader. Time is divided into terms, numbered 1, 2, 3, and so on. Each term has at most one leader.

• The leader sends regular heartbeats to the followers, a few times a second.
• If a follower hears nothing for a while (its election timeout, chosen at random from a range such as 150 to 300 ms), it assumes the leader is gone. It increases the term, becomes a candidate, votes for itself, and asks the others for their votes.
• Each server votes at most once per term, for the first candidate that asks (with one important condition, below).
• A candidate that gets votes from a majority becomes leader for that term, and starts sending heartbeats.
• If a server ever sees a message with a higher term, it learns it's out of date and immediately becomes a follower.

The random timeouts are a small, clever detail. If every follower waited exactly the same time, they'd all become candidates at once, split the votes, and nobody would win. With random timeouts, one usually times out first, asks for votes, and wins before the others even start.

And the higher-term rule is what makes stale leaders harmless. When the frozen old leader of term 3 wakes up and sends a heartbeat, the followers reply "we're on term 4 now," and it steps down. It can't get a majority to accept anything from an old term.

Raft: agreeing on a log

Electing a leader is only half of it. The point of consensus is usually to agree on a sequence of decisions: a log, much like the write-ahead log (lesson 1.4) or the Kafka log (lesson 4.1).

• Clients send requests to the leader. The leader appends each to its log and sends it to the followers.
• Once a majority have stored an entry, the leader marks it committed, applies it, and replies to the client. Committed entries are never lost, because any future leader must come from a majority, and at least one member of that majority has the entry.
• The voting condition: a server only votes for a candidate whose log is at least as up to date as its own. That's what guarantees a new leader already has every committed entry.

That's most of Raft. The paper covers the rest: how followers whose logs have diverged get corrected, how membership changes safely, how logs are compacted with snapshots.

What it costs

Every committed entry needs a round trip to a majority. Inside one region, that's a millisecond or two. Spread across regions, every decision pays a cross-region round trip (lesson 5.4's PACELC again). And a group can only make progress while a majority is reachable: 3 servers survive 1 failure; 5 survive 2. That's why consensus groups are kept small (3 or 5, never 4, because 4 survives no more failures than 3), and why they're used for small, important decisions rather than for all of an application's data.

Step 6 of 14 · 1 min

Where you'll find consensus

You'll rarely implement Raft or Paxos yourself, and you shouldn't: they're famously hard to get exactly right. You'll use systems built on them:

• etcd (Raft), which stores Kubernetes' cluster state, and is used for configuration and leader election in many systems.
• ZooKeeper (using a similar protocol called Zab), long used by Hadoop, HBase and older versions of Kafka.
• Consul (Raft), for service discovery and configuration.
• Kafka itself now uses a Raft-based protocol for its own metadata.
• Distributed databases such as CockroachDB, TiKV and Spanner run a separate consensus group for each range of data, so every write is agreed by a majority of that range's replicas (lesson 5.2's ranges, made reliable).

These systems typically offer exactly the building blocks this lesson described: leader election, leases with ever-increasing version numbers that work as fencing tokens, and a small, strongly consistent store for configuration and coordination.

Step 7 of 14 · 1 min

Do you need it?

Consensus is powerful, and also often not the right tool. Before reaching for it, ask whether the problem can be designed away:

• Make the operation idempotent (lessons 1.5 and 4.2), so doing it twice is harmless. Then "exactly one" relaxes to "at least one," which is far easier. Chirp's daily digest (lesson 2.1) relied on this: the lock reduced duplicates, and the unique constraint made them harmless.
• Let one existing system decide. A single database's row lock or unique constraint already gives you "exactly one" for anything stored in that database. Lesson 5.3's giveaway code needed nothing more.
• Avoid needing a leader. Many tasks can be split so each worker owns a partition (Kafka consumer groups, lesson 4.1), with the assignment managed by a system that already uses consensus.

Reach for etcd or ZooKeeper (or a managed equivalent) when you genuinely need a single, fault-tolerant decision-maker shared across many machines: choosing leaders for your own services, assigning unique IDs or partitions, or storing configuration that every server must agree on.

Step 8 of 14 · 1 min

From the real world: a switch that was half broken

Consensus algorithms are designed around clean failures: a node crashes, or a network link is cut. Cloudflare's control-plane outage on 2 November 2020 is a good look at what a partial failure does instead.

Cloudflare ran an etcd cluster (etcd uses Raft) that, among other things, tracked the health of its database clusters. A network switch in one rack partly failed: some traffic got through, some didn't, and the switch didn't fail in a way that triggered the usual failover to its partner. The etcd node in that rack, node 1, could no longer reliably reach the leader, node 3, but could still reach node 2. Node 2 could reach both.

Node 1 stopped hearing the leader's heartbeats, so it did what Raft says: it started an election, voting for itself. Node 2, which could still see the healthy leader, didn't vote for it. Node 1 timed out and tried again, and again. The cluster couldn't keep a stable leader long enough to accept writes.

The damage spread through what depended on etcd. Database management systems that used it couldn't confirm their primaries were alive, so they promoted replicas. A defect then required replicas to be rebuilt, and one authentication database, left without its usual read replica, became overloaded, slowing Cloudflare's API and dashboard. The switch recovered by itself within minutes; the knock-on effects lasted about six and a half hours.

Two lessons: partial failures (a node that can reach some peers but not others) are harder for consensus systems than crashes, and automated failover that trusts a coordination service inherits that service's bad days. Make sure "I can't confirm the primary is healthy" doesn't automatically mean "promote something," without checking other signals.

Step 9 of 14 · 1 min

Going deeper: PreVote and CheckQuorum

Plain Raft has a weakness that matches the shape of Cloudflare's incident: a node that's only partly cut off can repeatedly disrupt a healthy cluster. Raft's designers proposed two extensions for it, and etcd implements both.

In plain Raft, a follower that stops hearing heartbeats increases its term (Raft's election counter) and asks for votes. Any server that sees a higher term must accept it, and a leader that sees a higher term steps down. So a node with a flaky link, which keeps missing heartbeats, keeps forcing elections with ever-higher terms. Each one can knock a perfectly healthy leader off its seat, even though the rest of the cluster can see that leader fine.

• PreVote: before starting a real election, a node asks "would you vote for me in the next term?" without increasing its term. Others say yes only if they also haven't heard from a leader recently. A node that's merely cut off from the leader gets no's, stays a follower, and never disturbs anyone. When its link recovers, it catches up quietly.
• CheckQuorum: a leader that hasn't heard from a majority of the cluster within an election timeout steps down on its own. This stops a leader that has been isolated from the cluster from continuing to act as leader (and serving stale reads), and works together with PreVote so followers that can still hear a valid leader ignore disruptive election requests.

If you run etcd, ZooKeeper, Consul or a Raft library yourself, check whether these options are on. They're cheap, and they turn a whole class of "flaky node takes down the cluster" incidents into "flaky node is slightly behind."

Step 10 of 14 · 1 min

What Chirp changed

• Moved ID-generator leases to etcd, whose leases come with an increasing revision number, used as the fencing token.
• Generators stop issuing IDs if they haven't confirmed their lease recently, measured on a monotonic clock that keeps counting through pauses.
• Database failover (lessons 2.5 and 5.4) uses a 3-zone consensus group to choose leaders; storage rejects writes from older terms.
• Scheduled jobs use etcd's leader election, and every job is idempotent anyway.
• A design-review rule: "Who decides, and what happens if they freeze?" for anything described as "only one."

Step 11 of 14 · 1 min

In the simulation

The "incident day" epilogue includes a frozen leader: a database leader or a job runner stops for 30 seconds and then resumes. Without fencing, it writes stale data after a new leader has taken over, and the dashboard counts the damage. With term numbers or fencing tokens, its writes are rejected, and nothing changes.

Open the simulation →

Step 12 of 14 · 1 min

If you're here for an interview

Leader election and locks come up in designs with schedulers, job runners, ID generation or database failover. Strong answers:

• Say why it's hard: you can't tell slow from dead, processes can freeze, clocks drift.
• Use leases with fencing tokens, and explain that the protected resource checks the token.
• Use majorities: odd-sized groups (3 or 5), and only a majority can elect a leader.
• Use an existing system (etcd, ZooKeeper, a managed equivalent) rather than inventing a protocol, and know roughly how Raft elects a leader and commits entries.
• Prefer idempotency where it removes the need for "exactly one" altogether.

Step 13 of 14 · 2 min

Try these yourself

Step 14 of 14 · 1 min

Further reading

• Diego Ongaro and John Ousterhout, "In Search of an Understandable Consensus Algorithm" (the Raft paper), 2014. Unusually readable; the raft.github.io site has an interactive visualization.
• Martin Kleppmann, "How to do distributed locking," 2016, on leases and fencing tokens.
• Martin Kleppmann, Designing Data-Intensive Applications, Chapter 8 ("The Trouble with Distributed Systems") and Chapter 9 ("Consistency and Consensus").
• Leslie Lamport, "Paxos Made Simple," 2001.
• etcd documentation on leases, leader election and revisions.
• Cloudflare blog, "A Byzantine failure in the real world," November 2020.
• Diego Ongaro, Consensus: Bridging Theory and Practice (PhD thesis, 2014), section 9.6 (preventing disruptions when a server rejoins); etcd documentation on --pre-vote and Raft CheckQuorum.