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

10.7.1 said that strong consistency is paid for in round trips. These are the protocols that spend them:

• Two-phase commit, for all-or-nothing transactions across systems, and why it can get stuck.
• Raft, the consensus algorithm you should be able to explain step by step.
• Leader election.
• Distributed locks, the tool most often used wrongly in production. Its fix, the fencing token from 10.4, 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 all-or-nothing. But the costs explain why 2PC is rare in modern systems:

• It can get stuck. Suppose the coordinator crashes after the participants have said "prepared" but before they hear the decision. They sit there holding locks. They cannot commit or abort, because they promised to wait, and they cannot decide on their own. This is the classic "in-doubt transaction", and a person has to step in.
• Availability multiplies. The transaction only succeeds if every participant and the coordinator are all up. So the system is as unreliable as all of them combined, not as reliable as the best one.
• It holds locks across a network round trip, which slows everything down.
• Every participant must speak the protocol (XA). Most modern databases do not, and no message broker does.

Three-phase commit adds a step to reduce blocking under some failure models and is essentially unused in practice.

What replaced it:

• Across services: sagas, a series of local transactions, each with an undo step (10.8.4).
• For a database plus a message broker: the transactional outbox (10.8.4).
• Inside one distributed database: commit built on consensus. Spanner and CockroachDB still run 2PC, but every participant and coordinator is itself a Raft group. So no single machine failing can leave things stuck. They fixed 2PC by making every role survive failures, rather than dropping 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. Safety no longer depends on "the lock is right". It depends on "the storage refuses out-of-date writers", which can actually be enforced.

Practical guidance:

• Use locks freely for efficiency, to avoid doing the same work twice.
• Use them for correctness, to stop two writers changing the same data, only with fencing.
• Better still, design the lock away. Conditional writes, one writer per key, and operations that are safe to repeat remove most needs for a lock. Control over concurrency belongs where the data lives (9.5.2).

(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. It explains:

• why a 3-node etcd survives one failure but not two;
• why a 2-node cluster is worse than a 1-node cluster, since any single failure loses the majority;
• why adding nodes survives more failures but makes writes slower;
• why the smaller side of a split becomes read-only;
• why split brain cannot happen in a properly set up consensus system, but easily happens in a database where someone promotes a new leader by hand (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 a waste. 4 nodes still survive only 1 failure. A majority of 4 is 3, so losing 2 leaves 2, which is not a majority. And you pay for an extra machine and an extra acknowledgement on every write. A 2-node cluster is worse than a single node. Any failure leaves 1 of 2, which is not a majority, so the cluster stops. A single node alone at least keeps serving 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 real question is whether running it twice does harm. If the reconciliation writes adjustments, sends notifications or moves money, two runs at once can apply everything twice. A Redis lock cannot prevent that. The holder can freeze during garbage collection for longer than the lock's time limit, while the job is halfway through (section 3). A single Redis server also loses its locks when it fails over. And Redlock, the version across several Redis servers, is open to the same timing problem.

The correct solution, in layers, strongest first.

(1) Make the job safe to run twice. This is the main defence, and it stops you depending on any lock. Give each reconciliation run a key based on its date, such as recon-2026-07-22. Claim that key in one atomic step in the database, using a unique constraint. And make every adjustment it writes safe to repeat, using a natural key (10.4). Now a second run does nothing, whatever 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 defences, and you need both.

(a) Fencing. Claiming the job hands out a token, a number that only ever goes up, saved on the job's row. Every write the worker makes carries that token: progress updates, results, and calls to outside APIs where they support it. The database rejects any write with an out-of-date token (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.

Losing a whole availability zone:

• The queue and the database keep copies in several zones (10.5).
• Workers hold no state, so new ones start in the zones that survived.
• The consensus cluster keeps its majority, by design.
• Jobs that were running in the dead zone have their leases run out, and workers elsewhere pick them up.

That last point is the payoff of the whole design. Prove it in a practice drill, rather than assuming it works.

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.