You're building the control plane for a database spanning 3 availability zones: who is the current sole primary (accepting writes)? Can this shard-migration task start right now? What's the address of db-shard-7 in service discovery? Every one of these demands that all nodes see the same answer, and that the answer never splits into two under a network partition. This is distributed coordination, and the industry's standard answer is a consensus service like etcd / ZooKeeper / Chubby.
Concrete design goal: a 5-node strongly-consistent coordination service (etcd-like) —
2f+1=5, tolerate f=2 node failures and still accept writes; under a partition, never split-brain (never produce two committing leaders).
graph TD
C["Client
kubelet / db node"]
subgraph RSM["Consensus Group (Replicated State Machine)"]
L["Leader
term=7"]
F1["Follower A"]
F2["Follower B"]
F3["Follower C"]
F4["Follower D"]
end
SM["State Machine KV
apply committed log"]
C -->|"1. write"| L
L -->|"2. AppendEntries
replicate log"| F1
L -->|"2."| F2
L -->|"2."| F3
L -->|"2."| F4
F1 -.->|"3. ack"| L
F2 -.->|"3. ack"| L
L -->|"4. majority ack -> commit"| SM
SM -.->|"5. apply then reply"| C
classDef leader fill:#2a1530,stroke:#ff7ab6,color:#e8eef5
classDef follower fill:#1a2530,stroke:#64c8ff,color:#e8eef5
classDef sm fill:#1a1a30,stroke:#ffb450,color:#e8eef5
classDef client fill:#0e2030,stroke:#5eead4,color:#e8eef5
class L leader
class F1,F2,F3,F4 follower
class SM sm
class C client
A write enters the leader's log, replicates to a majority (3/5), and only then commits and applies to the state machine — "majority acknowledgment" is the bedrock of the whole system's safety
Every write is serialized into a replicated log; each node applies it to its state machine in identical order and thus reaches identical state — the Replicated State Machine model. The leader orders; followers merely replicate. Consensus (Raft / Paxos / ZAB) solves exactly one problem: given that nodes crash and networks drop and partition, make all live nodes reach an irrevocable agreement on "what entry N of the log is."
Core trade-off: a majority quorum trades "unavailable once more than half the nodes are down" for "any two decisions necessarily intersect, so they can't contradict." This is the mathematical bedrock of a CP system; the price is that availability is capped by the quorum.
Principle. Consensus must satisfy Agreement (all nodes agree on one value), Validity (the agreed value was actually proposed), and Termination (eventually decides). The mechanism is majority intersection: among 5 nodes, any two majorities (each ≥3) must share at least 1 node, and that node "remembers" the prior decision, preventing two conflicting committed values. That is exactly why 2f+1 nodes tolerate only f failures — you need a live majority to form a new decision while guaranteeing it intersects the historical one.
| Algorithm | Strengths | Sacrifice |
|---|---|---|
| Multi-Paxos | Earliest, most general, out-of-order commit | Brutal to understand/implement correctly; the paper omits engineering details |
| Raft | Strong leader, contiguous log, understandability-first | In-order commit only; leader is the write bottleneck |
| ZAB | ZooKeeper's atomic broadcast + recovery | Coupled to ZK semantics, not general-purpose |
# Raft log replication core (leader side, pseudocode)
def on_client_write(cmd):
entry = LogEntry(term=current_term, cmd=cmd)
log.append(entry) # append to own log first
replicate_to_followers(entry) # parallel AppendEntries
# wait for a majority (including self)
if acked_count(entry) >= majority: # 3/5
commit_index = entry.index # advance commit point
apply_to_state_machine(entry) # only now does it take effect
return OK
# no majority -> block/timeout, NEVER commit
# Follower receiving AppendEntries: reject stale term, never go backwards
def on_append_entries(req):
if req.term < current_term: return REJECT # stale leader
if not log_matches(req.prev_index, req.prev_term): return REJECT # log gap
log.append(req.entries); return ACK
Core trade-off: a monotonic term plus majority election guarantees at most one committable leader at any instant; the price is that the cluster is write-unavailable during elections (a few hundred ms).
Principle. Each election produces a monotonically increasing term (ballot in Paxos, epoch in ZAB). A candidate wins only with a majority of votes, and majority intersection guarantees no two leaders can be elected in one term. Under a partition, a minority side (≤2 nodes) can't reach a majority, so it can neither elect a new leader nor commit any write; the majority side elects a leader with a higher term. Even if the old leader is still alive and still thinks it's leader, its AppendEntries carrying the old term is REJECTED by everyone — this is how split-brain is killed mathematically: you can have two "self-proclaimed leaders," but only one can assemble a majority to commit.
# Storage rejects a stale leader via a fencing token (monotonic check is key)
current_token = 0
def write(payload, token):
global current_token
if token < current_token: # smaller = stale leader / stale lease
raise StaleLeaderError # reject! even if it still thinks it's leader
current_token = token
persist(payload)
Core trade-off: a consensus-based lock (ZK/etcd) is slower than Redis Redlock but buys provable uniqueness; a pure-TTL lock inevitably produces two owners under GC/network delay.
Principle. The classic illusion: client A acquires a lock (TTL=10s); while working it hits a sudden 15s stop-the-world GC pause; the lock expires, B acquires it and starts writing; A wakes up, unaware it lost the lock, and keeps writing → two clients hold the lock and write the same resource at once. TTL can never plug this hole, because there's always a gap between "check the lock" and "use the lock." The only correct fix: the lock service issues a monotonically increasing fencing token; the client attaches the token to every downstream operation, and storage rejects any token smaller than the largest it has seen — A wakes up with a stale token and its write is bounced.
| ZooKeeper ephemeral-sequential | etcd lease | Redis Redlock | |
|---|---|---|---|
| Uniqueness guarantee | Strong (consensus + session) | Strong (Raft + lease) | Disputed (Kleppmann's critique) |
| Fencing token | zxid built-in | revision built-in | None built-in; must roll your own |
| Failure detection | Session heartbeat, auto-release on disconnect | Lease keepalive | TTL expiry |
| Latency | Higher (goes through consensus) | Higher (goes through consensus) | Low (in-memory) |
# ZooKeeper style: ephemeral-sequential node = natural fair lock + fencing
def acquire_lock(zk):
my = zk.create("/lock/req-", ephemeral=True, sequential=True) # yields req-000007
while True:
kids = sorted(zk.children("/lock"))
if my == kids[0]: # my number is smallest -> hold lock
return seq_num(my) # the number IS the fencing token (monotonic)
prev = predecessor(kids, my) # watch only the node before me -> no herd
zk.watch(prev) # predecessor released/session lost -> wake me
wait()
Core trade-off: quorum read is safe but slow; lease read is fast but bets on the clock; follower read scales but may read stale. Pick one, depending on whether you truly need linearizability.
Principle. Intuitively "the leader has the latest data, just read its local memory" — wrong. A leader may have just been deposed by a majority and a new leader elected, while it doesn't know yet (a partition isolated it). Reading its local state returns a stale value, breaking linearizability. The correct approach: before reading, the leader must confirm it is still leader — Raft's ReadIndex: record the current commit index, do one round of heartbeats to confirm leadership still holds with a majority, then return the value at that index.
# Raft ReadIndex: confirm leadership before reading, avoiding stale-leader reads
def linearizable_read(key):
idx = commit_index # record current commit point
if not confirm_leadership_via_quorum(): # one heartbeat round; a majority must reply
raise NotLeaderError # deposed -> reject, forward to real leader
wait_until(applied_index >= idx) # wait for state machine to catch up
return state_machine.get(key) # what we read now cannot be stale
AppendEntries and allow several in flight, lifting throughput from hundreds to tens of thousands of writes/sec.Other frequent follow-ups: (1) Why not use 2PC for coordination? (Coordinator single point; participants block; not fault-tolerant.) (2) How do you place 5 nodes across 3 AZs to survive one AZ dying? (3) What's the real difference between Raft and Paxos? (4) How do watches guarantee no missed events (etcd revision + resumable streaming)?
What consensus safety truly needs is not "a majority" but "a new decision's quorum is guaranteed to meet an old decision's quorum" — a majority is merely the simplest sufficient condition for intersection (any two sets each >N/2 must intersect).
Flexible Paxos's insight: as long as the election quorum Q1 and replication quorum Q2 satisfy |Q1| + |Q2| > N (guaranteeing intersection), it's safe. So you can configure them asymmetrically: shrink the replication quorum (e.g. only 2 acks to commit among 5 nodes) to cut write latency, at the cost of a larger election quorum (needing 4). For "write-frequent, election-rare" loads this is a net win — moving cost off the hot path (every write) onto the cold path (occasional election).
Corollary: you can also do "grid quorums" and other topologies so cross-region replication needs only a local majority plus a few remote nodes, optimizing geographic latency. Essentially it exposes the "intersection" constraint as an explicit tuning knob.
Key: after any single AZ dies, the remaining nodes must still form a majority (3/5).
Cost optimization: if the third AZ should hold only 1 node, use a "witness / no-data arbiter" — it only votes and stores no full data, cheaply buying the odd vote and tie-breaking power.
A longer lease doesn't help: you don't know how long the pause lasts; whatever you set can be beaten by a longer pause; and the longer the lease, the longer everyone else waits when a real crash happens, wrecking availability.
Renewal doesn't help: during a GC pause the entire process is frozen, so it can't execute renewal heartbeats. By the time it wakes, the lock has long changed hands elsewhere.
Root cause: no matter how you tune the timing, there's always a window between "the client checks am I still the lock holder" and "the client actually writes," and the pause can land precisely in that window. It's a timing problem, not a duration problem — lengthening times just makes the window harder to hit, not gone.
Why a fencing token cures it: it moves the verdict from "the client's unreliable self-perception" to "the storage's monotonic validation." The awakened old client arrives with token=33, but storage has already raised its watermark because the new holder wrote with token=34, so 33 < 34 is rejected outright. Correctness no longer depends on any timing assumption — even if the client pauses for an hour, its old token can never get through. This turns a "timing problem" into a "monotonic-ordering problem."
Why default to quorum read: only by confirming "I am still leader right now and have seen all committed writes" can you guarantee linearizability. Reading local state directly = betting you weren't quietly deposed, and Jepsen proved that reads backwards values.
When follower reads are safe: (1) the business can accept bounded staleness (e.g. "at most 5s stale"), giving up linearizability for throughput; or (2) the follower uses a read-index mechanism — ask the leader for the current commit index, wait for local apply to catch up to that index, then return; this yields a "no-earlier-than-request-time" linearizable read at the cost of waiting to catch up (TiKV's approach). Purely "read local follower memory" is always non-linearizable.
Why leader-lease reads skip round-trips: the leader holds a time-bounded leadership lease; as long as it's within the lease and the system guarantees "no new leader is elected before the old lease expires" (a new election must wait out the old lease), the leader can be sure it's still the sole leader and read locally without a per-read heartbeat. Cost: correctness depends on bounded clock drift — a frozen VM or a big NTP jump can break it, so you leave a safety margin (compute the lease more conservatively than the theoretical value).
Why adding machines doesn't raise throughput, or even lowers it: within a single Raft/Paxos group all writes serialize through the same leader, which is the absolute bottleneck. Adding followers doesn't share writes — instead each log entry must replicate to more nodes, the leader's outbound network and fsync pressure grow, and the majority threshold rises (7 nodes need 4 acks, so tail latency is dragged by the slowest). That's why consensus groups usually stop at 5-7 nodes: beyond that you only add fault tolerance, not performance.
Multi-Raft's breakthrough: since one group can't scale, run thousands of groups. Split the key space into many ranges/shards; each shard is an independent Raft group with its own leader, log, and majority. Spread the shard leaders across different physical nodes → write throughput scales with the shard count, and one machine is simultaneously leader of some groups and follower of others.
New costs: atomic operations across shards need an extra distributed-transaction layer (2PC over Raft groups); shard splitting/merging/rebalancing becomes a new source of complexity; and heartbeats for thousands of Raft groups flood the network, so you need Raft group heartbeat coalescing (batching multi-group heartbeats between the same pair of nodes into one). CockroachDB and TiKV exemplify this route.