Design a Consensus System (Raft / Paxos / etcd)

Raft's first job: elect exactly one leader at a time, even with arbitrary failures.

Roles
At any moment, each node is in one of three roles: follower (passive, applies entries from leader), candidate (running for election), or leader (replicating entries).

Term
A monotonically increasing integer. Each new election bumps the term. Stale leaders (with older terms) detect this when they hear from a higher-term node and step down.

Election triggering
A follower has an election timeout (randomized between 150-300ms typically). If it doesn't hear from the leader within the timeout, it becomes candidate, increments its term, votes for itself, and requests votes from peers.

Voting rules
A node votes for at most one candidate per term. Candidate wins if it gets a majority. Vote granted only if:

• Candidate's term >= voter's term.
• Voter hasn't already voted this term.
• Candidate's log is at least as up-to-date as voter's (specific rule: highest term in last entry, then highest index).

The "up-to-date" rule prevents a candidate with stale log from winning - a critical safety property.

Split-vote
Multiple candidates start elections simultaneously, none gets a majority. All time out, increment term, retry. Without randomization, this could loop forever.

The fix: randomized election timeouts (150-300ms range). One node will time out first, become candidate, and most likely win before others start. If split happens, it usually resolves in 1-2 rounds.

Election latency

• Best case: leader fails, one follower times out at ~150ms, wins immediately. Total ~150-200ms.
• Worst case (split-vote, 2-3 rounds): ~500ms-1.5s.

This is the unavailability window for writes during leader failure. Tune timeouts based on requirements:

• Lower timeout: faster failover, more false positives (network blip → election).
• Higher timeout: stable, but slow recovery.

Pre-vote optimization
A node about to start an election first asks "would you vote for me?" without incrementing term. If it can't get majority, it stays follower. Avoids unnecessary term bumps when the network is partitioned.

Leader transfer
For planned operations (rolling restart, drain a node), the current leader can voluntarily transfer leadership: nominate a follower, ensure it's caught up, send TimeoutNow. The follower starts an election immediately and wins. Faster and cleaner than waiting for timeout.

Once leader is elected, replicating writes is mostly mechanical - but the safety invariants are subtle.

AppendEntries
Leader receives client write. Appends to its log. Sends AppendEntries RPC to followers with the new entry plus a "previous index/term" hint.

Each follower checks: does my log have an entry at the previous index with that term? If yes, append. If no, reject (log inconsistency).

On rejection, leader decrements the next-index for that follower and retries. The protocol marches the leader's view of the follower's log back until it finds a matching point, then catches up forward.

Commit rule
An entry is committed when:

• It's been replicated to a majority of replicas (including leader).
• The leader's term equals the entry's term (i.e., entry was created in current term).

The second condition is critical. It prevents a subtle bug: a leader from term T1 replicates entry at index I but doesn't commit before crash. New leader from term T2 may overwrite it. Even if T2 sees the entry replicated to a majority, it can't commit it just on majority alone - it must commit something from term T2 first, which transitively commits the older entry.

Why the second condition matters (Figure 8 in the Raft paper)
Without it, a leader could crash, a new leader from a later term could overwrite a committed entry. Real systems hit this. The condition is non-negotiable.

Heartbeats
The leader sends empty AppendEntries periodically (every 50-100ms typically) to maintain its leadership and prevent followers from starting elections.

Pipelining
Naive: leader sends entry I, waits for ack, sends I+1. High-latency.
Optimized: leader sends I, I+1, I+2 in flight. Acks come back asynchronously. Throughput limited by network bandwidth, not RTT.

Modern Raft implementations (etcd-raft, hashicorp/raft, tikv/raft) all pipeline.

Batching
Group N small writes into one AppendEntries RPC. Reduces fsync overhead (one fsync per batch, not per entry). Increases throughput dramatically (10x+).

Trade-off: latency rises with batch size. Tunable.

Flow control
Leader tracks each follower's match-index (highest replicated). If a follower lags, leader sends only a window of entries; doesn't drown it.

If a follower is far behind (out of log retention), leader sends InstallSnapshot instead of replaying the whole log.

The log can't grow forever. Snapshots compact old entries.

The snapshot mechanism
Each replica periodically (every N entries or T minutes) writes a snapshot of its state machine to disk. The snapshot includes the last applied index and term.

Once snapshotted, log entries below the snapshot index can be discarded. Disk usage stays bounded.

Snapshot installation
A new node joining (or one too far behind to catch up via log) receives the leader's snapshot via InstallSnapshot RPC. It loads the snapshot, then applies any subsequent log entries.

InstallSnapshot is chunked (snapshots can be GBs). Each chunk acked. The receiver writes to a temp file; on completion, atomically swaps in.

Snapshot frequency vs replay cost

• Frequent snapshots (every 1K entries): smaller log retention, more snapshot overhead.
• Infrequent (every 1M entries): larger log, longer cold-start replay.

Default in etcd: 100K entries. Tune based on entry size and crash-recovery time SLA.

Streaming snapshots vs blocking
Naive: pause the state machine, write snapshot, resume. Blocks writes during snapshot.
Modern: copy-on-write or fork() to take a consistent point-in-time view; serialize asynchronously while writes continue. Standard in production implementations.

Snapshot to object storage
For backup / DR, periodically push snapshots to S3. Restore: bootstrap a new cluster from snapshot. RPO = snapshot interval.

Compaction lag
If a follower is taking longer than expected to apply, leader's log can't be compacted past the slowest follower's match-index. Slow followers prevent compaction; if persistent, kick them out and let them rejoin via snapshot install.

Linearizable reads are the read-side equivalent of consensus on writes. Done naively, every read needs a quorum round-trip. Optimizations cut this dramatically.

The naive approach: quorum read
For a linearizable read, the leader exchanges a heartbeat with a majority of followers, then serves the read from its state. The heartbeat confirms the leader is still legitimate (no newer leader has taken over).

Cost: 1 round-trip per read. Acceptable for low-throughput coordination services; expensive at high read rates.

Read index optimization (Raft paper §6.4)

1. Leader records its current commit-index as read-index.
2. Leader confirms its leadership by exchanging heartbeats with majority.
3. Once leader's apply-index >= read-index, serve the read from local state.

Why this is linearizable: any read served sees all writes committed before the read started. The heartbeat round prevents reading from a deposed leader.

Cost: 1 round-trip but batchable. Many reads can share one heartbeat round.

Leader leases
The leader holds a lease that's valid for a time window (~10s typically). While the lease is valid, the leader is guaranteed to be the unique leader (no other node could have won an election). Reads can be served without any round-trip.

Lease invariant: leases are granted by followers as part of voting. Followers won't vote for a new leader until their grant of the lease expires. As long as the leader's clock isn't ahead of follower clocks by more than the lease, the lease is safe.

Clock skew danger
Leases assume bounded clock skew. If clocks skew by more than the lease window, two leaders could believe they hold the lease simultaneously. NTP is critical; cloud VMs typically skew <1ms in practice.

Some implementations use a hybrid (lease + occasional verification heartbeat) for safety margin.

Stale read trade-offs
Some clients accept stale reads (read-your-writes is enough). They can read directly from any replica without leader involvement. Most consensus systems offer both APIs - linearizable (slow) and stale (fast).

Read scaling
With leader leases, reads scale to the leader's CPU/memory ceiling - a single Go process can serve 100K+ linearizable reads/sec. Beyond that, follower reads (lease-based on followers, more complex) extend the ceiling.

Adding or removing nodes safely is harder than it looks.

The naive approach (broken)
Decree the new membership, switch over. Problem: during the switchover, a partition could create two majorities - one from old config, one from new config - electing two leaders simultaneously.

Joint consensus (Raft paper §6)
Phase 1: leader proposes a "joint" configuration that includes both old and new members. While in joint, decisions require majorities from BOTH old AND new sets - so no single config can act unilaterally.

Phase 2: once joint config is committed, leader proposes the new (final) config. Once committed, joint is retired. Old-only members can be removed.

Two configuration changes; safe under any partition pattern.

Single-node changes (alternative)
Change membership one node at a time. With odd cluster sizes, a single add/remove can't create two disjoint majorities (the majority overlaps).

Simpler than joint; supported by etcd's raft library. Most production systems use this for routine ops.

Pitfall: the new node hasn't caught up
If you add node N6 to a 5-node cluster, and N6 has empty log, you've added a non-functional voter. If two old nodes fail before N6 catches up, the cluster loses majority.

Solution: add as a non-voter (learner) first. N6 catches up via InstallSnapshot + log replication. Once caught up, promote to voter via a second config change.

etcd, TiKV, CockroachDB all use the learner pattern.

Removing the leader
The leader can step down by transferring leadership to another node first, then removing itself. Cleaner than letting it crash mid-removal.

Cluster too small
2-node clusters offer no fault tolerance (1 failure = no majority). 1-node "clusters" are not consensus, just a single point of failure dressed up.

3 is the minimum for production. 5 is the standard for important services. Beyond 7, the throughput cost (more replicas to ack) usually exceeds the marginal availability gain.

Consensus is expensive. Use it where you need it; avoid it where you don't.

Where you need consensus

• Service discovery / config (etcd, Consul, ZooKeeper).
• Distributed locks / leader election.
• Metadata for distributed databases (Spanner, TiDB control plane).
• Small, frequently-read, infrequently-written state.
• The control plane of distributed systems generally.

Where consensus is the wrong tool

• High-throughput data stores. Consensus per write doesn't scale to 1M writes/sec. Use eventual consistency + CRDTs (Cassandra), per-shard consensus (Spanner), or async replication (Postgres + replicas).
• Large state. Consensus replicates the full log to every node. 10TB state via Raft is painful; use sharded consensus (each shard is its own small Raft group).
• Cross-region writes. Inter-region consensus latency is 50-200ms per write. Most apps can't tolerate this. Use eventual consistency cross-region; consensus within region.

The CAP framing in practice
CAP says: during partition, choose consistency or availability. Consensus systems choose consistency - the minority side stops accepting writes. The majority side continues.

For etcd / control planes: this is correct. A split-brain configuration system corrupts everything downstream.

For user-facing systems: availability often wins. Cassandra serves both sides of a partition, reconciles later. The trade-off is application-specific.

Sharded consensus: the production answer for scale
Spanner and CockroachDB shard data by key range. Each shard has its own 3-5 node Raft group. 1000 shards = 1000 independent consensus groups. Aggregate throughput scales with shards.

Per-shard ops are linearizable (within shard). Cross-shard ops use 2PC over Raft groups for ACID transactions. Complex but achievable.

The "we just need a leader" pattern
Many use cases need consensus only to elect a single coordinator (job scheduler, primary among replicas). Once elected, the coordinator does the work without per-op consensus.

etcd / ZooKeeper provide this primitive. Almost every distributed system uses one of them for leader election rather than implementing Raft directly.

Don't roll your own
Implementing Raft correctly is famously hard. Use etcd-raft, hashicorp/raft, or atomix/copycat as a library, or run etcd / ZooKeeper / Consul as a service. The bugs in custom implementations are usually safety bugs - silent data loss under specific failure timings.