System Design: Consensus and Leader Election

1. The leader sends the entry to followers.
2. Each follower appends it, after checking that its log matches the leader's at the previous entry (the log matching check). If it does not match, the follower rejects it and the leader backs up and resends earlier entries until the logs agree. This repairs followers that missed entries.
3. When the entry is stored on a majority of nodes, it is committed. The leader applies it to its state machine and replies to the client.
4. Followers learn the commit point from later messages and apply the entry too.

Committed entries are durable and will survive any leader change, because a node with a missing committed entry cannot win an election. Entries that were not yet committed when a leader fails may be lost or may survive, and the client must handle the uncertain outcome, typically by retrying with an idempotency key (see Distributed Transactions).

What a partition does

In a five-node cluster split three and two, the side with three nodes can still elect a leader and commit entries. The side with two cannot reach a majority, so it makes no progress. An old leader stranded on the minority side keeps believing it is the leader until it hears a higher term, but it cannot commit anything, because it cannot get a majority acknowledgement. Raft therefore chooses consistency over availability in a partition, which is the CAP trade-off in action. See CAP and Consistency Models.

Paxos, stated accurately

Paxos is the older, foundational consensus family, introduced by Leslie Lamport. Basic Paxos agrees on a single value using proposers, acceptors and learners, with numbered proposals and two phases: a prepare phase to claim a proposal number and discover any value already accepted, and an accept phase to get that value accepted by a majority. Multi-Paxos repeats this to agree on a sequence of values, typically with a stable leader that skips the first phase in the normal case, so it ends up resembling Raft's steady state.

Paxos and Raft give comparable safety guarantees for crash failures. Raft was designed for understandability and a complete specification, while Paxos is subtle to implement and has many variants. Both handle crash faults with a majority; neither handles malicious (Byzantine) nodes, which need different protocols and larger quorums.

ZooKeeper and etcd: consensus as a service

Most teams should not implement consensus themselves. They use a small, well-tested cluster that exposes it through a simple API.

These stores give you primitives rather than raw consensus: watches that notify you of changes, ephemeral keys that vanish when a client's session ends, and compare-and-set operations. A typical leader election recipe: every candidate tries to create the same key or node; the one that succeeds is the leader; the others watch it, and when the leader's session expires the key disappears and a new election happens.

Keep the data small and the write rate low: these are coordination stores, not general databases.

Leases: leadership with an expiry

Heartbeats and elections handle crashes, but how does a leader know it has been replaced? A lease is a time-limited grant of authority: "you are leader until time T unless you renew". If the leader cannot renew, because it is partitioned or paused, it must stop acting as leader when the lease runs out. Meanwhile, the rest of the cluster waits at least that long before granting a new lease.

Leases let a leader serve reads without a quorum round trip and make locks clean themselves up after crashes. But they rely on bounded clock drift: the holder and the granter must measure roughly the same duration. A holder that pauses for longer than the lease and then wakes up without noticing is the root of the next problem.

Split-brain and fencing tokens

Consider this scenario with a distributed lock that has a lease.

1. Client A acquires the lock and starts writing to a storage service.
2. A suffers a long pause (a garbage collection pause, a stalled VM, a slow network). Its lease expires.
3. Client B acquires the lock and writes to the storage service.
4. A wakes up, still believing it holds the lock, and writes stale data over B's.

A checks its own clock and finds the lease valid or does not check at all, because the pause happened between the check and the write. No amount of care inside A can fully close that gap. The fix has to be in the resource being protected.

A fencing token is a number that increases with every lock grant. The lock service gives A token 33 and later B token 34. Every write to the storage service includes the token, and the storage service remembers the highest token it has seen and rejects any request with a lower one.

Raft's terms play a similar role inside the protocol: a deposed leader's messages carry an old term and are rejected. When you build on top of etcd or ZooKeeper, you can use a revision or transaction id as the token. The storage side must support checking it, so this works best when you control the resource or it offers conditional writes (compare-and-set on a version).

Choosing and operating consensus

A few practical guidelines:

• Use a managed or proven system. Writing your own consensus is a multi-year project with subtle failure modes. Prefer etcd, ZooKeeper, a managed service, or a database that embeds Raft or Paxos.
• Place nodes across failure domains. Three nodes in three availability zones survive the loss of a zone. Three nodes in one rack do not survive the loss of the rack. Cross-region groups survive more but pay in commit latency (every commit waits for a quorum round trip across regions).
• Tune timeouts to your network. Too short and transient latency triggers needless elections; too long and failover takes longer than users tolerate.
• Monitor leader changes. Frequent elections indicate network trouble, overloaded nodes or poor timeout settings, and each election pauses writes.
• Test with partitions and pauses. Fault-injection tests that cut links and freeze processes find the bugs in your use of the lock or lease.

Read it as a ladder of rising cost. Most background work belongs at the top, where a duplicate is harmless because the work is idempotent.

In practice: a job scheduler that needs exactly one active leader on Kubernetes

The situation

A B2B billing platform runs a scheduler service that triggers around 300 jobs an hour (illustrative): invoice generation, payment retries, usage roll-ups and customer emails. Ten engineers deploy it on Kubernetes with PostgreSQL as the system of record and RabbitMQ carrying work to workers. To survive pod restarts the scheduler runs as three replicas. Right now every replica evaluates every schedule, so some jobs fire twice, and an email goes out twice about once a week (assumption). An earlier attempt to fix it, "only replica 0 schedules", means that when that pod restarts, nothing is scheduled for several minutes. The business wants no duplicate invoices and a failover gap under roughly 30 seconds.

Starting architecture

All three pods can read and fire any schedule, and nothing arbitrates between them.

Options considered

The decision

Do not write your own data into the cluster's etcd. Use the Lease API, which is designed for exactly this.

Rollout and verification

Start with election in shadow mode: every pod logs "would fire" so you can compare the logs against the old duplicates. Turn on leadership, then epoch enforcement. Before production, run three fault tests: delete the leader pod, block its route to the API server, and freeze it with a stop signal. Dashboards and alerts: leader_transitions_total, lease renewal latency, missed_schedule_seconds (target under the failover budget), fenced_write_rejected_total (any increase is worth a look but proves fencing works), and a count of duplicate fires, which must stay at zero.

The 2 a.m. failure

The page reads: "Scheduler leader changed 5 times in 10 minutes". A node under CPU pressure starves the leader, which fails to renew in time, and the old leader later resumes and writes.

Detection: the transitions counter plus a non-zero fenced_write_rejected_total. First three checks: kubectl get lease on the scheduler lease to see the holder, transition count and renew time; kubectl describe the leader pod and its node for resource requests and pressure conditions, plus the pod's CPU throttling metrics; and the logs for "failed to renew lease" messages with API latency. Mitigation: raise the pod's CPU request so renewals are not starved, and temporarily lengthen the lease duration. Follow-up fix: guaranteed resources for the scheduler, an alert on transition rate, and a fault-injection test that freezes the process every release.

Architect's notes

• Ask first whether you need a leader at all. If jobs are idempotent, running on all replicas with a database claim is simpler than election.
• Measure the failover budget the business really needs. Sub-5-second failover usually costs more than 30 seconds does (tighter timeouts, more risk of flapping).
• Fence at the resource. The lease keeps duplicates rare; the epoch check keeps them harmless.
• Own the coordination layer. If the cluster team owns the API server, agree what load and latency you can expect from it.
• Test pauses, not just crashes. Crash tests pass easily; stalls expose the real bugs.
• Defer on purpose. Multi-region leadership adds cross-region quorum latency; do it only with a real requirement.

Common mistakes

• Using an even-sized cluster or only two nodes. Two nodes cannot form a safe majority after one fails, and four tolerate no more failures than three. Run three or five.
• Using wall-clock time to pick a leader or decide a lock is free. Clocks drift and jump. Use a consensus-backed store and logical terms.
• Locks without fencing. A paused client resumes and overwrites newer data. Add fencing tokens or conditional writes at the resource.
• Treating etcd or ZooKeeper as a general database. Large values and heavy write rates degrade the whole cluster. Store small coordination data only.
• Spreading a quorum over a high-latency link without noticing. Every write then pays the cross-region round trip. Measure commit latency before choosing placement.
• Ignoring the minority side during a partition. Clients connected to the minority will see errors or stale reads. Define what they should do, and test it.
• Assuming a committed client write is certain after a timeout. A timed-out request may or may not have committed. Retry with idempotency keys.

Key takeaways

• Consensus gives durable agreement. Nodes agree on a value that never flips, tolerating crashes and lost messages.
• Majorities overlap. That overlap is the reason any later decision sees an earlier one; clusters use odd sizes of 3 or 5.
• Raft separates concerns. Randomised-timeout leader election, majority-committed log replication, and an election restriction for safety.
• Terms are logical clocks. Messages from lower terms are rejected, so stale leaders are demoted quickly.
• Partitions favour consistency. The majority side makes progress; the minority side stops, which is the CAP trade-off in practice.
• Paxos and Raft are siblings. Both handle crash faults with a majority; neither tolerates malicious nodes.
• Use etcd or ZooKeeper-style services. They expose consensus as watches, ephemeral keys and compare-and-set, and are for small coordination data.
• Fence your locks. Leases handle crashes but not pauses; monotonically increasing tokens checked by the resource make stale leaders harmless.

Practice exercises

Exercise 1 — Starter (30 min): For clusters of 3, 5 and 7 nodes, calculate the majority size and how many failures each tolerates. Then explain in two sentences why a 4-node cluster is a poor choice compared with a 3-node or 5-node one.

Exercise 2 — Intermediate (2 hours): Draw a timeline for a five-node Raft cluster where the leader (term 3) is partitioned away together with one follower. Show which side can elect a leader, what term they reach, what happens to client writes sent to the old leader, and what the old leader's log looks like when the partition heals. Then use the interactive visualisation at the Raft website to check your answer.

Exercise 3 — Advanced (half day): Design a job scheduler where exactly one instance should run each scheduled job at a time, using etcd or ZooKeeper for leader election. Specify the election recipe, lease duration and renewal interval, what the scheduler does if it loses its session mid-job, and how fencing tokens or conditional writes protect the job-state store from a paused former leader. Write three fault-injection tests (partition, process pause, coordination-store restart) and the expected behaviour for each.

Related reading

• Raft consensus website - the paper, an interactive visualisation, and a list of implementations.
• CAP and Consistency Models - why a partitioned minority must stop or serve stale data.
• Replication - leader-follower setups, failover and split-brain from the data side.
• IDs, Clocks and Ordering - why logical clocks and terms replace wall-clock time.