10.7.2 — Consensus: 2PC, Raft, Leader Election & Distributed Locks

10.7.1 said strong consistency is bought with round trips. This page shows the protocols that spend them: two-phase commit (atomic transactions across systems — and why its blocking problem matters), Raft (the consensus algorithm you should be able to explain step by step), leader election, and distributed locks — the mechanism most often used incorrectly in production, with a fix (10.4's fencing tokens) that must be part of every answer.

1. Two-phase commit: atomicity across systems

The problem: a transaction spans two databases (or a database and a message broker), and you need all-or-nothing across both. 2PC coordinates it: ^{⚑What is two-phase commit and what are its limitations? [EQ-489b]}

1. Prepare — the coordinator asks every participant "can you commit?" Each does the work, makes it durable, takes locks, and answers yes (a promise it can commit even after a crash) or no.
2. Commit/abort — if all said yes, the coordinator writes its decision durably and tells everyone to commit; any no means abort everywhere.

The guarantee is real atomicity; the costs are why 2PC is rare in modern architecture: it's blocking — if the coordinator crashes after participants prepared but before the decision arrives, participants sit holding locks, unable to commit or abort (they promised), and cannot decide alone — the classic "in-doubt transaction" needing operator intervention; it couples availability (the transaction succeeds only if every participant and the coordinator are up — availability is the product, not the max); it holds locks across a network round trip, throttling throughput; and it needs participants that implement the protocol (XA), which most modern datastores and all message brokers do not. Three-phase commit adds a step to reduce blocking under some failure models and is essentially unused in practice.

What replaced it: for cross-service work, sagas — sequences of local transactions with compensating actions (10.8.4); for the database-plus-broker case, the transactional outbox (10.8.4); and within a single distributed database, consensus-based commit (Spanner and CockroachDB run 2PC over Raft groups, so no single coordinator failure blocks anything — the fix is making every role fault-tolerant rather than abandoning the protocol).

2. Raft: consensus you can explain

Consensus is getting a set of nodes to agree on a value (or on the next entry in a log) despite crashes and message loss. Raft (Ongaro & Ousterhout, 2014) was explicitly designed to be understandable where Paxos is not, and it's what etcd, Consul, TiKV, and CockroachDB use. Three sub-problems: ^{⚑Explain the Raft consensus algorithm. [EQ-1099b]}

Leader election. Every node is a follower, candidate, or leader, and time is divided into terms (monotonically increasing numbers — a logical clock, 10.3). A follower that hears no leader heartbeat within a randomized election timeout becomes a candidate, increments the term, and requests votes. A node grants at most one vote per term, and a candidate winning a majority becomes leader. Two details carry the correctness: randomized timeouts (150–300 ms typically) make simultaneous candidacies unlikely and break ties quickly; and majority means at most one leader per term can exist (two majorities always intersect, and the intersection node voted once) — that intersection property is the heart of the algorithm.

Log replication. Clients send commands to the leader, which appends to its log and sends AppendEntries to followers. Once a majority has stored an entry, the leader marks it committed, applies it to its state machine, and returns to the client; followers apply committed entries in order. Because every entry carries its term and index, and followers reject entries that don't match their expected predecessor, logs converge to identical sequences.

Safety. The property that makes it usable: a committed entry is never lost. It's guaranteed by the election restriction — a candidate can only win if its log is at least as up-to-date as the majority that votes for it, so any node with a committed entry blocks the election of a node lacking it. This is why Raft can promise safety always, liveness when the network behaves (10.3's FLP contract).

Figure 1 — Raft's commit rule. The leader appends and replicates; once a majority has durably stored the entry it is committed and acknowledged. Majorities are what make the system tolerate f failures with 2f+1 nodes — and why a slow or dead minority never stalls writes.

Paxos, in one paragraph: the original (Lamport, 1989/1998) solves the same problem with proposers, acceptors, and learners, and a two-phase prepare/accept exchange with ballot numbers. It's more general (Multi-Paxos handles log replication; Flexible Paxos generalizes quorum choices) and considerably harder to implement correctly, which is precisely why Raft exists. Practically: understand Raft, recognize Paxos, and know they provide the same guarantees.

Quorum arithmetic: a cluster of 2f+1 nodes tolerates f failures. Hence 3 nodes (survive 1), 5 nodes (survive 2). Even-numbered clusters are wasteful — 4 nodes still tolerate only 1 while requiring 3 for a majority. And every additional node makes writes slower (bigger quorum, more messages), which is why consensus clusters are small (3–7) and hold coordination data rather than bulk data.

3. Leader election and distributed locks

Both are consensus applications, and both are where teams get hurt.

Leader election — one node performs a role at a time: running a cron job (9.9.7's N-workers problem), owning a partition, coordinating a rebalance. Implementations: a consensus store's lease (etcd lease with keep-alive, ZooKeeper ephemeral node, Kubernetes Lease objects) or a database row with a TTL (UPDATE leader SET holder=?, expires=? WHERE expires < now() — a conditional write, 9.5.4). The essential property: leases expire, so a crashed leader is replaced automatically — and therefore a live but paused leader can lose its lease without knowing.

Distributed locks — the same mechanism used for mutual exclusion, and the mechanism most often used wrongly. The critical truth: ^{⚑How do distributed locks work and where do they fail? [EQ-1101b]}

A distributed lock cannot guarantee mutual exclusion by itself, because a lock holder can pause (GC, VM freeze, network partition) past its lease expiry and wake up believing it still holds the lock.

The lock service cannot distinguish paused from dead (10.3's FLP), so it hands the lease to someone else — and now two processes think they hold it. The fix is fencing (10.4): the lock hands out a monotonically increasing token; every write to the protected resource carries it; the resource rejects any token lower than the highest it has seen. Correctness moves from "the lock is right" to "the storage refuses stale writers" — which is enforceable. Practical guidance: use locks for efficiency (avoid duplicate work) freely; use them for correctness (mutual exclusion on state) only with fencing, or better, design the lock away — conditional writes, per-key single-writer ownership, and idempotent operations remove most needs for one (9.5.2's "concurrency control belongs at the state"). (The Redis Redlock debate — Martin Kleppmann's critique and Salvatore Sanfilippo's response — is exactly this argument; know that it exists and that the fencing point is its core.)

4. The expert lens

Consensus is expensive, so use it for metadata, not data. Every committed entry requires a majority round trip, so consensus clusters store cluster membership, configuration, leadership, locks, and partition maps — kilobytes that change rarely — while bulk data lives in systems that consult them. This is why etcd holds Kubernetes' state rather than your application's rows, and why a well-designed system has a small consensus core and a large non-consensus periphery.

"Just use a distributed lock" is usually the wrong instinct. The three better answers, in order: (1) make the operation idempotent so concurrent execution is harmless (10.4); (2) route by key so one owner naturally handles one entity, making exclusion structural rather than negotiated (10.6); (3) push the check into the storage as a conditional write/CAS, which is atomic by construction. Reach for a lock when none of those fit — and then add fencing.

Understanding the majority rule explains a dozen operational behaviors. Why a 3-node etcd survives one failure but not two; why a 2-node cluster is worse than a 1-node cluster (any single failure loses quorum); why adding nodes increases fault tolerance but decreases write speed; why a partitioned minority goes read-only; why split brain is impossible in a properly configured consensus system but easy in a manually-promoted database (10.5). One property — two majorities always intersect — generates all of it.

Next: 10.8.1 — the messaging layer where most production distributed systems actually live: queues, streams, delivery, and the patterns built on them.

Recall

• 2PC: prepare (participants promise, durably) → commit/abort (coordinator decides). Real atomicity, but blocking (coordinator crash leaves in-doubt transactions holding locks), availability multiplies, locks held across the network, and needs XA support. Replaced in practice by sagas and the outbox; modern distributed SQL runs 2PC over Raft groups so no role is a single point of blocking.
• Raft = leader election + log replication + safety. Terms as logical clocks; randomized election timeouts; a node votes once per term so a majority yields at most one leader; entries commit once a majority stores them; the election restriction (candidate's log at least as up-to-date) is what makes committed entries unlosable. Contract: safety always, liveness when the network behaves. Paxos: same guarantees, harder — understand Raft, recognize Paxos.
• Quorum arithmetic: 2f+1 nodes tolerate f failures (3→1, 5→2); even sizes waste a node; more nodes = more fault tolerance but slower writes ⇒ consensus clusters stay small (3–7) and hold metadata, not bulk data.
• Leader election = a lease that expires (etcd/ZooKeeper/K8s Lease, or a conditional DB row) — which also means a paused leader can lose it silently. Distributed locks cannot guarantee mutual exclusion alone (paused holder vs FLP): correctness requires fencing tokens rejected at the resource. Better: make the operation idempotent, route by key for single ownership, or use a conditional write.
• Lens: consensus for metadata; "just use a lock" is usually third-best; majorities intersect explains split-brain prevention, quorum loss, and why 2-node clusters are worse than 1.

Self-test: Why is 2PC blocking, and what exactly is an in-doubt transaction? Walk Raft's three sub-problems and name the property preventing two leaders per term. Why does a paused lock holder break mutual exclusion, and what fixes it? How many failures does a 5-node cluster tolerate, and what does adding two more nodes cost?

Quiz Bank

FoundationalExplain two-phase commit, its guarantee, and its four costs.

Protocol: phase 1 (prepare) — the coordinator asks each participant whether it can commit; each performs the work, makes it durable, holds locks, and replies yes (a binding promise it can still commit after a crash) or no. Phase 2 (commit/abort) — the coordinator durably records the outcome (all-yes → commit, any-no → abort) and instructs everyone.

Guarantee: genuine atomicity across independent systems. Costs: (1) Blocking — if the coordinator dies after prepares and before the decision, participants are in doubt: they promised, so they may neither commit nor abort unilaterally, and they hold locks until an operator or a recovered coordinator resolves them. (2)

Availability multiplies — the transaction needs every participant and the coordinator available, so overall availability is the product of the parts, not the best of them. (3) Locks held across network round trips, sharply reducing throughput under contention. (4)

Ecosystem — participants must implement XA-style protocols, which most modern datastores and essentially no message brokers do. What's used instead: sagas with compensation for cross-service workflows, the transactional outbox for the database-plus-broker case (10.8.4), and — inside distributed SQL engines — 2PC layered over consensus groups, so every participant and the coordinator are themselves fault-tolerant and nothing blocks on a single node's death.

FoundationalWalk through Raft: the three sub-problems and the safety argument.

(1) Leader election. Nodes are followers, candidates, or leaders; time is divided into terms (increasing integers — a logical clock). A follower that misses heartbeats for a randomized timeout (≈150–300 ms) becomes a candidate, increments the term, votes for itself, and requests votes. Each node grants one vote per term; a candidate with a majority becomes leader. Randomization prevents perpetual split votes; the one-vote rule plus majority means at most one leader per term (any two majorities share a node, which cannot vote twice).

(2) Log replication. Clients send commands to the leader, which appends them and issues AppendEntries; each entry carries its term and index, and followers reject entries whose predecessor doesn't match — a consistency check that forces logs to converge. Once a majority has durably stored an entry, the leader commits it, applies it to the state machine, and responds to the client; followers apply committed entries in the same order, so all state machines are identical.

(3) Safety. The key promise is that a committed entry is never lost, enforced by the election restriction: a candidate only receives a vote if its log is at least as up-to-date as the voter's, so any node holding a committed entry refuses to elect a node missing it — and since a committed entry is on a majority, every possible winning majority includes at least one such node. Result:

safety unconditionally; liveness whenever a majority can communicate (10.3's FLP-compatible contract).

AppliedWhy can't a distributed lock guarantee mutual exclusion, and how do fencing tokens fix it?

Because the lock service must decide liveness by timeout, and a timeout cannot distinguish a crashed holder from a paused one (10.3's FLP). Concretely: process A acquires a 30-second lease; A suffers a 45-second stop-the-world GC pause (3.6.9) or its VM is frozen for migration; the lease expires and the service grants it to B; A resumes, still believing it holds the lock, and writes. Two writers, mutual exclusion violated — and no lock implementation can prevent it, because the violation occurs after the lock check succeeded.

Fencing: the lock service issues a monotonically increasing token with each grant (A gets 33, B gets 34); every write to the protected resource includes its token, and the resource rejects any token lower than the highest it has already accepted. A's post-pause write carries 33, the storage has seen 34, so the write is refused — enforcement moved from the lock (which can only advise) to the resource (which can decide). Requirement: the protected resource must support the check (a conditional write on a stored token, an object-storage precondition, a database column).

Better still, where possible: eliminate the lock — make operations idempotent, give each key a single owner via partitioning, or use a conditional write/CAS that is atomic by construction (9.5.2).

InterviewWhy do consensus clusters have 3, 5, or 7 nodes, and what does adding nodes actually change?

Consensus requires a majority for both election and commit, so a cluster of 2f+1 nodes tolerates f failures: 3 tolerates 1, 5 tolerates 2, 7 tolerates 3. Even sizes are wasteful: 4 nodes still tolerate only 1 (a majority needs 3, so losing 2 leaves 2 — no quorum), while costing an extra machine and an extra ack per write; and a 2-node cluster is worse than a single node, because any failure leaves 1 of 2 — no majority — so it's unavailable, whereas one node alone at least serves until it dies.

Adding nodes changes three things: fault tolerance rises (good), write latency rises (a bigger majority means waiting for more acknowledgments, and the slowest member of the required majority sets the pace), and message volume grows (the leader replicates to everyone). That trade is why clusters stay at 3–7 and why consensus systems store coordination metadata (membership, configuration, leases, partition maps — kilobytes, changing rarely) rather than bulk application data. Related operational facts worth stating: read scaling doesn't come from more voters (reads still need the leader for linearizability, or lease-based leader reads), which is why systems add non-voting learners/observers for read replicas and geographic reach without slowing the quorum.

StaffYour team needs exactly one worker to run a nightly reconciliation across a 12-instance fleet. They propose a Redis lock. Evaluate, and design the correct solution.

Evaluate the proposal: a Redis SET NX lock is the reflexive answer, and for efficiency purposes (avoiding wasteful duplicate work) it's acceptable — running the job twice merely burns compute. But the question is whether duplicate execution is harmful: if reconciliation writes adjustments, sends notifications, or moves money, then two concurrent runs can double-apply, and a Redis lock cannot prevent that (section 3's paused-holder problem — the holder can GC-pause past its TTL while the job is mid-flight; single-node Redis also loses locks on failover, and Redlock's multi-node variant remains vulnerable to the same timing argument).

The correct solution, layered by strength: (1) Make the job idempotent — this is the primary defense and removes the dependence on any lock: key each reconciliation run by date (recon-2026-07-22), claim that key atomically in the database (a unique constraint), and make every adjustment it writes idempotent by natural key (10.4); now a second run is a no-op regardless of what the lock did. (2)

Use a real lease with fencing where exclusion matters — if some step genuinely cannot be idempotent (an external notification, a file export), acquire a lease from a consensus store (etcd/ZooKeeper/Kubernetes Lease) whose token is carried into the protected operation and checked by the resource. (3)

Better: remove the election entirely — schedule the job outside the fleet: a Kubernetes CronJob, a cloud scheduler, or a queue message with a unique key, so exactly one execution is created by construction rather than negotiated by twelve competitors (9.9.7's recommendation). (4)

Observability: emit a run record with start/end/claimed-by, alert if the run doesn't happen (silent non-execution is the failure mode nobody notices until month-end) and if it happens twice. The design principle to leave with the team: locks make duplicate work unlikely; idempotency makes duplicate work harmless — build the second and treat the first as an optimization, and prefer an architecture where the scheduler creates one job to one where twelve instances race for the right to be the scheduler.

Flashcards

Flash2PC's problem

Coordinator crash after prepare = in-doubt participants holding locks, unable to decide. Availability multiplies. Replaced by sagas/outbox, or run over Raft groups.

FlashRaft in three parts

Leader election (terms, randomized timeouts, one vote per term, majority) · log replication (append, majority ack ⇒ commit) · safety (election restriction: up-to-date log required to win).

FlashWhy majorities work

Any two majorities intersect ⇒ at most one leader per term, and a committed entry is on every possible winning quorum. Explains split-brain prevention and quorum loss.

FlashQuorum math

2f+1 tolerates f (3→1, 5→2). Even sizes waste a node; 2-node is worse than 1. More nodes = more tolerance, slower writes ⇒ keep 3–7, store metadata only.

FlashDistributed locks

Can't guarantee exclusion alone (paused holder vs FLP). Fix: fencing tokens rejected at the resource. Better: idempotency, per-key ownership, conditional writes.

Scenario Drill

DrillDesign the coordination layer for a distributed job-processing platform: 200 workers, jobs must run exactly once, some jobs are long-running (hours), workers can crash or be preempted mid-job, and the platform must survive losing a whole availability zone. Specify what uses consensus, what doesn't, and how a crashed mid-job worker is handled safely.

What uses consensus (small, metadata only): a 5-node etcd/Raft cluster spread across 3 AZs holds coordination state — worker registry, queue-shard assignments, the leader of the scheduler component, and configuration. Five nodes across three zones survives a full-AZ loss (losing 2 of 5 still leaves a majority — the quorum arithmetic doing exactly what it's for), while remaining small enough that write latency stays low.

What does not use consensus (everything with volume): job payloads and state live in a durable queue plus a database — bulk data through a Raft log would be both slow and unnecessary (section 4's metadata rule). Exactly-once, honestly: the platform provides at-least-once delivery plus idempotent execution (10.4) — a job's effects are keyed by job_id so re-execution converges, and the job record's state machine (queued → running → done | failed | abandoned) is advanced by conditional writes so two workers cannot both mark it running. Promising literal exactly-once would be a lie the moment a job calls an external API (10.4).

Long-running jobs and crashed workers — the interesting part: a worker claims a job by conditional write (UPDATE jobs SET owner=?, lease_expires=now()+90s WHERE id=? AND status='queued') and then renews the lease every 30 seconds while working. If the worker crashes or is preempted, renewal stops, the lease expires, and a reaper (or any worker) re-claims the job — this is leader election per job, and leases are what make crash recovery automatic. The danger is the section 3 scenario: a worker that paused (long GC, node freeze) rather than died can wake up after its lease was reassigned and continue writing.

Two defenses, both required: (a) fencing — the claim issues a monotonically increasing token stored on the job row; every write the worker makes (progress updates, results, external calls where the API supports it) carries the token, and the database rejects writes bearing a stale one (10.4); (b)

self-checking workers — before each significant side effect, the worker verifies its lease is still valid and its token still current, aborting if not (a cheap read that turns most zombie writes into a clean abort rather than a rejected write). Preemption specifically (spot instances) gets a graceful path: the platform handles the termination signal by releasing the lease immediately and checkpointing progress (9.9.7's shutdown discipline), so re-execution resumes rather than restarts — long jobs should checkpoint anyway, keyed idempotently so partial work is reusable.

AZ loss: queue and database are multi-AZ replicated (10.5); workers are stateless and re-provision in surviving zones; the consensus cluster keeps quorum by construction; in-flight jobs in the dead AZ have their leases expire and are re-claimed elsewhere — which is the whole design's payoff, and the property to demonstrate in a game-day exercise rather than assume.

The summary line for the design doc: consensus holds the small facts that must be agreed; leases plus fencing tokens turn "which worker owns this job" into an automatically recoverable, safely-rejected-on-zombie decision; and idempotent job effects mean the worst outcome of any failure is repeated work, never wrong work.