Consensus and leader election

Article.

How does a cluster make sure exactly one node acts as leader, and that a deposed leader cannot still write, when processes pause, networks partition and clocks drift? That one question carries most of what “distributed systems fundamentals” means in an interview: partial failure, timeouts as failure detectors, clocks, quorums, Raft, leases and fencing tokens.

You rarely implement consensus. You use it every day, through etcd under Kubernetes, KRaft in Kafka 4, RabbitMQ quorum queues, Consul, ZooKeeper, and failover managers such as Patroni and Redis Sentinel. The senior skill is knowing what those systems guarantee and what your code must still do, because the classic production bug is two nodes that both believe they are the leader.

Replication modes and consistency models are in Replication, sharding and consistency models; CAP and ACID in CAP, ACID, BASE, PACELC.

Partial failure: why this is hard

On one machine a call returns or the process crashes. Across a network, a request with no reply can mean any of:

• the request was lost, or the node is dead;
• the node is alive but slow, or paused (garbage collection, swapping, a stalled VM);
• the node processed it and the reply was lost.

The caller cannot tell these apart. A timeout is the only failure detector there is, and it is a guess. That is the practical content of two theory results interviewers like to name:

• Two generals: over a link that can lose messages, no protocol lets two parties both be certain they agree. Every acknowledgement needs its own acknowledgement.
• FLP (1985): in a fully asynchronous system with even one crashing process, no deterministic algorithm guarantees consensus terminates. Real systems escape by assuming partial synchrony: messages usually arrive within some bound, so timeouts eventually work. They stay safe when the bound is violated, and only lose liveness (no leader for a while).

Choosing a timeout

A timeout below the worst pause causes false failovers; one far above it makes real failures slow to detect. Set it above the p99 response time plus the worst GC or stall pause, and prefer detectors that adapt:

• Heartbeats with a fixed timeout: etcd’s defaults are a 100 ms heartbeat and a 1,000 ms election timeout.
• Phi-accrual (Cassandra, Akka): track the distribution of heartbeat intervals and output a suspicion level instead of a yes/no, so a normally jittery link is not declared dead.
• Gray failure: a node that answers health checks but fails real work (a full disk, a broken dependency) defeats both. Health checks must test the path real requests take.

Clocks lie

Leases, timeouts and “last write wins” all depend on time, and a server has two kinds of clock:

• Wall clock (time.time()): synchronised by NTP, which can step it backwards or forwards. Ordering events by it across machines is wrong by the clock offset, typically milliseconds and occasionally seconds.
• Monotonic clock (time.monotonic()): never goes backwards, but means nothing on another machine. Use it for every timeout and TTL inside one process.

The ordering failure, simulated with three processes where B’s clock runs 40 ms behind. A sends an order to B, which then sends a payment to C:

A sends order      wall=1000.005  lamport=1
B receives         wall=999.970  lamport=2
B sends payment    wall=999.975  lamport=3
C receives         wall=1000.020  lamport=4

By wall clock, B received the order before A sent it. A sort on wall timestamps puts the payment before the order it pays for.

Logical clocks

• Lamport clock: a counter; increment on each event, send it with each message, and on receipt set max(local, received) + 1. If event a caused event b, then L(a) < L(b), as in the output above. The converse does not hold: a smaller number does not mean “caused”.
• Vector clock: one counter per node. Two versions where neither vector dominates the other were concurrent, so a store can detect a conflict instead of silently dropping one write (Dynamo, Riak).
• Hybrid logical clock (CockroachDB, YugabyteDB): physical time plus a logical counter, so timestamps stay close to wall time but still respect causality.
• TrueTime (Spanner): GPS receivers and atomic clocks give a time interval with a known uncertainty of a few milliseconds; each commit waits out that uncertainty before becoming visible, which buys externally consistent transactions at the price of commit latency.

For “order the events of one entity”, the practical answer is usually none of these: route the entity’s events through one partition of a log (a Kafka partition per key) or one sequencer, which gives a total order per key.

Quorums: why majorities

A decision counts once a majority of the N nodes accepts it. Any two majorities share at least one node, so two conflicting decisions cannot both win. The arithmetic of cluster size:

Four nodes tolerate the same single failure as three, with one more machine to lose and one more acknowledgement to wait for. In a 2-2 split neither side has a majority, so the whole cluster stops accepting writes. Hence odd sizes, and usually 3 or 5: every write waits for a majority, so a larger group is slower, not faster.

Raft, at the mechanism level

Raft (2014) is the consensus algorithm in etcd, Consul, KRaft, CockroachDB and RabbitMQ quorum queues. Each node is a follower, candidate or leader.

         election timeout          majority of votes
follower ───────────────▶ candidate ───────────────▶ leader
   ▲                          │                        │
   └── sees a higher term ────┴──── sees higher term ──┘

1. Terms are numbered epochs. Every message carries the sender’s term; a node that sees a higher term steps down to follower immediately.
2. Election: a follower that hears no heartbeat within its election timeout (randomised, 150-300 ms in the paper, so candidates rarely collide) increments the term, votes for itself and sends RequestVote.
3. Voting: each node grants one vote per term, and only to a candidate whose log is at least as up to date as its own. So a new leader always has every committed entry.
4. Replication: the leader appends client commands to its log and sends AppendEntries. An entry is committed once a majority stores it; only then is it applied and acknowledged to the client.
5. Reads: serving reads from the leader’s memory is not enough, because a deposed leader in a minority partition may not know it. Linearizable reads either confirm leadership with a majority first (ReadIndex) or rely on a lease that assumes bounded clock drift.

What it costs: every write needs a round trip to a majority, so a 5-node group spread over regions commits only when the second-closest follower answers. That is why consensus groups usually stay within one region, and why they hold small, critical state (membership, configuration, leases) rather than application data.

Leases, and the leader that does not know it is deposed

Most application code gets leadership as a lease from one of these systems: “you are leader until T”. The failure, from the simulation below: worker A holds a 1-second lease, then pauses for 2 seconds (a GC pause, a swapped-out page, a stalled VM). The lease expires, B acquires it and writes. A wakes up still believing it is leader, and writes too:

--- fencing OFF
t= 0.0s  A acquired lease, token 33
t= 0.0s  A pauses 2.0s (GC, swap, VM steal)
t= 1.0s  B acquired lease, token 34
t= 1.0s  B writes with token 34: ok
t= 2.0s  A writes with token 33: ok
t= 2.0s  final owner: A

Checking “do I still hold the lease?” just before writing does not fix it: the pause can happen between the check and the write. The process cannot protect itself, because it cannot observe its own pause.

Fencing tokens: the storage says no

The lease service hands out a number that increases with every grant (etcd’s revision, ZooKeeper’s zxid or sequential znode number, a Postgres sequence). The writer sends it with every write, and the storage rejects any token lower than one it has already seen:

UPDATE job
   SET owner = :me, fence = :token
 WHERE id = :id AND fence <= :token;   -- 0 rows → you are stale

--- fencing ON
t= 0.0s  A acquired lease, token 33
t= 0.0s  A pauses 2.0s (GC, swap, VM steal)
t= 1.0s  B acquired lease, token 34
t= 1.0s  B writes with token 34: ok
t= 2.0s  A writes with token 33: REJECTED (stale token)
t= 2.0s  final owner: B

The guarantee moves from “only one process thinks it is leader” (impossible) to “only the newest leader’s writes land” (enforceable). It requires the resource to check the token; a third-party API that cannot is protected only by idempotency keys. This is also the core of the case against Redlock: Redlock and the Kleppmann Critique.

Where you meet this in practice

• Kubernetes leader election (kube-controller-manager defaults): 15 s lease, 10 s renew deadline, 2 s retry. A controller that misses its renew deadline must exit; one that is paused cannot, so controllers rely on writes being idempotent.
• Patroni demotes a primary that can no longer renew its leader key, and can use a hardware or software watchdog to reset a node that is too stuck to demote itself. With asynchronous replicas, a failover can still lose acknowledged commits; synchronous_mode limits promotion to a synchronous standby.
• Redis Sentinel: an old primary isolated on the minority side keeps accepting writes, which are discarded when it rejoins as a replica. min-replicas-to-write 1 with min-replicas-max-lag 10 makes it refuse writes once it cannot see a replica, bounding the loss window.
• Kafka with unclean.leader.election.enable=true lets an out-of-sync replica become leader, trading lost messages for availability; see Kafka: Log Compaction, Rebalancing, and Advanced Topics.

Leadership through the database

A common pattern is one Postgres row or advisory lock as the “leader” flag. Two traps:

• A session advisory lock is held as long as the connection looks alive. A leader host that vanishes without closing its TCP connection keeps the lock until the server notices; with Linux keepalive defaults (7,200 s idle plus 9 probes 75 s apart) that is about 2 hours 11 minutes without a leader. Set tcp_keepalives_idle, tcp_keepalives_interval, tcp_keepalives_count and tcp_user_timeout on the server.
• The opposite failure is split brain: the leader’s connection drops, Postgres frees the lock, a new leader takes it, and the old process keeps working. Advisory locks: PostgreSQL Advisory Locks.

The robust form is a row with an epoch column, which is lock and fence in one: become leader with UPDATE leader SET holder = :me, epoch = epoch + 1 ..., and make every side effect conditional on epoch = :mine.

Where it breaks in production

• Election storms: timeouts set tighter than GC pauses or disk stalls, so leaders are deposed while healthy. Watch etcd_server_leader_changes_seen_total; more than a few an hour is a problem.
• Slow disks look like dead nodes: etcd fsyncs its log before acknowledging, and a etcd_disk_wal_fsync_duration_seconds p99 above about 10 ms makes heartbeats late. Consensus nodes need dedicated, fast disks.
• Lost quorum is a write outage: with 2 of 3 etcd members down, the Kubernetes API can serve nothing that needs a write; pods keep running but nothing can be scheduled or changed.
• Two schedulers both run the nightly job: a lease without fencing, or a lock keyed on Python’s per-process salted hash().
• Events in the wrong order: sorted by wall-clock timestamps from different hosts. Alert on clock offset (node_timex_offset_seconds from node_exporter, or chrony’s tracking).

The trade not taken

• TTL locks without fencing (Redlock, a Redis SET NX PX): fine for efficiency, such as avoiding duplicate work that is harmless if it happens; wrong for correctness, because nothing stops the paused holder.
• Two-phase commit for agreement across services: a coordinator crash between the phases leaves participants holding locks until it returns, so it blocks exactly when you need it not to. Sagas and outboxes are the usual answer: Data consistency across services.
• Running your own Raft: correct implementations take years of testing (Jepsen has found bugs in most systems it has tested). Use etcd or the database you already run; the epoch-row pattern above covers most application-level leader election without new infrastructure.

Related

• Replication, sharding and consistency models - replication modes, CAP, PACELC, quorums for data
• Redlock and the Kleppmann Critique - why TTL locks need fencing
• PostgreSQL Advisory Locks - Postgres locks as leader election
• Replication, Sentinel and Cluster - Redis Sentinel and Cluster failover
• MongoDB — Replica Sets and Sharding - MongoDB elections
• Timeouts, retries and backoff - choosing timeouts