System Design ·

Consensus Algorithms Explained: Raft, Paxos, and How Distributed Systems Agree

A deep-dive into distributed consensus: why FLP impossibility and network partitions make agreement hard, how Paxos works and why it's notoriously difficult to implement correctly, how Raft simplifies the problem with leader election and log replication, and when you actually need consensus versus simpler coordination approaches.

Consensus Algorithms Explained: Raft, Paxos, and How Distributed Systems Agree

You have three database replicas. One is the primary. The primary crashes mid-write. Two replicas have seen the write; one has not. A new primary needs to be elected, and every surviving node needs to agree on which one it is and what the last committed state was. If they disagree, you have a split-brain. If they take too long to agree, your service is unavailable. If they elect a node missing committed writes, you lose data.

This is the consensus problem. It looks like a coordination detail until it fails in production, and then it reveals that you were carrying a fundamental bet about distributed system behavior that turned out to be wrong.

Understanding consensus algorithms at the level of how they actually work, not just what they guarantee, is the difference between using etcd or ZooKeeper correctly and cargo-culting them into a configuration that breaks in ways you do not expect.

Why Consensus Is Hard

The intuition is that coordinating across multiple nodes is just a matter of sending messages and collecting votes. The reality is that three fundamental problems make this much harder than it looks.

Network Partitions and the CAP Theorem

A network partition is when nodes in your cluster cannot communicate with each other, but both subsets are still running. Neither subset knows whether the other crashed or is just unreachable. If you require availability (both sides keep serving requests), you risk each side making conflicting decisions. If you require consistency (only one side serves requests), you sacrifice availability. CAP is not a choice you make once at the start of a project. It is a constraint that surfaces during every partition event.

Consensus algorithms sit firmly on the CP side of this tradeoff. They require a quorum (majority) of nodes to agree before any decision is committed. A partition that isolates a minority segment causes that segment to stop making progress rather than risk a conflicting decision. This is the right call for anything involving committed state, but it means your system will be unavailable to that minority segment until the partition heals.

Clock Skew and Timing Assumptions

You cannot rely on physical clocks for ordering events in a distributed system. Two nodes whose clocks differ by 200ms can disagree on which operation happened first. Consensus algorithms solve this by using logical ordering (term numbers, ballot numbers, sequence numbers) rather than wall clock timestamps. The consequence: every consensus protocol you will encounter uses a monotonically increasing counter to identify rounds or eras, and any message from a prior era is rejected, regardless of when the physical clock says it arrived.

The FLP Impossibility Result

Fischer, Lynch, and Paterson proved in 1985 that in a purely asynchronous distributed system, no deterministic consensus algorithm can guarantee termination (liveness) in the presence of even a single process failure. This is FLP impossibility.

What it means practically: you cannot build a consensus algorithm that is simultaneously safe, live, and tolerant of failures in a fully asynchronous network. Every real consensus protocol resolves this by adding timing assumptions. Paxos and Raft assume that eventually the network will be well-behaved enough for messages to get through and for a stable leader to emerge. They do not guarantee this will happen, but they guarantee that if it does happen, the system will make progress and any progress made will be correct. Correctness (safety) is always maintained. Liveness is a best-effort under timing assumptions.

This is why you see timeouts and election randomization in every consensus implementation. They are heuristics that make the timing assumptions hold in practice, not theoretical guarantees.

Paxos: The Original Algorithm

Paxos was described by Leslie Lamport in 1989 (published 1998) and is the theoretical foundation for most consensus work that followed. Understanding it is worth the effort because it clarifies exactly what problem consensus is solving and why the solution has the shape it does.

How Paxos Works

Paxos operates in two phases. There are three roles: proposers (nodes that want to get a value decided), acceptors (nodes that vote), and learners (nodes that learn the decided value). In practice, every node plays all three roles.

Phase 1: Prepare

A proposer chooses a proposal number n (unique and higher than any it has used before) and sends Prepare(n) to a majority of acceptors. Each acceptor that receives Prepare(n) responds with a promise: “I will not accept any proposal numbered less than n.” If the acceptor has already accepted a proposal, it includes that proposal in its response. If the proposer receives promises from a majority, it moves to Phase 2.

Phase 2: Accept

The proposer sends Accept(n, v) to a majority of acceptors, where v is the value from the highest-numbered previous proposal it received in Phase 1 responses, or its own value if none were received. Each acceptor accepts the proposal unless it has already promised to ignore proposals below a higher number. If a majority accepts, the value is decided.

The invariant is subtle: if a value v was decided in some earlier round, any new proposer will discover v during Phase 1 (because any majority it contacts will overlap with the majority that accepted v) and will be forced to propose v again rather than a different value.

Why Paxos Is Hard to Implement

Single-decree Paxos (agreeing on one value) is understandable. Multi-Paxos (the variant you actually need to build a replicated log) introduces leader election, log gaps, instance numbering, and reconfiguration, none of which are covered by the original paper.

The gaps Paxos leaves unspecified:

  • How does a new leader discover the full committed log from previous leaders?
  • What happens when an acceptor misses some Accept messages, creating log gaps?
  • How do you change cluster membership without violating safety?
  • How do you handle the case where a leader completes Phase 1 but crashes before Phase 2, leaving acceptors in a state where they have promised but not accepted?

Every production implementation of Paxos (Chubby, Spanner, Cassandra’s lightweight transactions) fills these gaps differently and adds significant complexity beyond the base algorithm. Lamport himself noted that Paxos as described does not specify enough to build a complete system. This is the core reason Raft was designed.

Raft: Designed for Understandability

Raft was published by Diego Ongaro and John Ousterhout in 2014 with the explicit goal of being more understandable than Paxos. It solves the same problem but decomposes it into three subproblems that can be reasoned about separately: leader election, log replication, and safety.

Leader Election

Raft uses a term-based system. Time is divided into terms, each identified by a monotonically increasing integer. Each term begins with an election. If a candidate wins, it serves as leader for the rest of the term. If no candidate wins (split vote), the term ends without a leader and a new election begins.

Nodes start as followers. Each follower maintains an election timeout (randomized between 150ms and 300ms in the original paper). If a follower does not receive a heartbeat from a leader before the timeout fires, it assumes the leader has failed, increments its term, transitions to candidate, votes for itself, and sends RequestVote RPCs to all other nodes.

A node grants a vote if:

  1. The candidate’s term is at least as large as the voter’s current term.
  2. The voter has not already voted in this term.
  3. The candidate’s log is at least as up-to-date as the voter’s log (determined by comparing the term and index of the last log entry).

The third condition is the log completeness check. It ensures that a candidate missing committed entries cannot win an election, which is the core of Raft’s safety argument.

If a candidate receives votes from a majority (including its own vote), it becomes leader and immediately sends heartbeats to all nodes to prevent new elections. Randomized timeouts make it unlikely that two nodes start elections simultaneously, which avoids split votes in most cases.

Log Replication

Once a leader is elected, it handles all client requests. Each request is appended to the leader’s log as a new entry with the current term number and a sequence index. The leader sends AppendEntries RPCs to all followers in parallel, carrying the new entry (or just a heartbeat with no new entries).

A follower accepts the entry if:

  • The term in the RPC is at least as large as the follower’s current term.
  • The follower’s log contains an entry at the previous index with the previous term (the consistency check).

Once the leader receives acknowledgment from a majority of nodes (including itself), the entry is considered committed. The leader includes the commit index in subsequent AppendEntries RPCs so followers know which entries they can apply to their state machines.

Here is a TypeScript model of the core log replication data structures and the leader’s append path:

type LogEntry = {
  term: number;
  index: number;
  command: unknown;
};

type AppendEntriesRequest = {
  term: number;
  leaderId: string;
  prevLogIndex: number;
  prevLogTerm: number;
  entries: LogEntry[];
  leaderCommit: number;
};

type AppendEntriesResponse = {
  term: number;
  success: boolean;
  matchIndex?: number; // highest log index confirmed replicated
};

class RaftLeader {
  private log: LogEntry[] = [];
  private commitIndex = 0;
  private nextIndex: Map<string, number> = new Map();  // per follower
  private matchIndex: Map<string, number> = new Map(); // per follower

  constructor(
    private nodeId: string,
    private currentTerm: number,
    private peers: string[],
    private sendRpc: (
      peer: string,
      req: AppendEntriesRequest
    ) => Promise<AppendEntriesResponse>
  ) {
    // After winning election: initialize nextIndex to leader log length + 1
    for (const peer of peers) {
      this.nextIndex.set(peer, this.log.length + 1);
      this.matchIndex.set(peer, 0);
    }
  }

  async appendEntry(command: unknown): Promise<boolean> {
    const entry: LogEntry = {
      term: this.currentTerm,
      index: this.log.length + 1,
      command,
    };
    this.log.push(entry);

    const acks = await Promise.allSettled(
      this.peers.map((peer) => this.replicateToPeer(peer))
    );

    const successes =
      1 + // leader counts itself
      acks.filter((r) => r.status === "fulfilled" && r.value).length;

    const majority = Math.floor((this.peers.length + 1) / 2) + 1;
    if (successes >= majority) {
      this.commitIndex = entry.index;
      return true;
    }
    return false;
  }

  private async replicateToPeer(peer: string): Promise<boolean> {
    const nextIdx = this.nextIndex.get(peer) ?? 1;
    const prevIndex = nextIdx - 1;
    const prevEntry = prevIndex > 0 ? this.log[prevIndex - 1] : null;

    const req: AppendEntriesRequest = {
      term: this.currentTerm,
      leaderId: this.nodeId,
      prevLogIndex: prevIndex,
      prevLogTerm: prevEntry?.term ?? 0,
      entries: this.log.slice(nextIdx - 1),
      leaderCommit: this.commitIndex,
    };

    const resp = await this.sendRpc(peer, req);

    if (resp.term > this.currentTerm) {
      // Discovered higher term: step down to follower
      this.currentTerm = resp.term;
      throw new Error("stale term: stepping down");
    }

    if (resp.success && resp.matchIndex !== undefined) {
      this.matchIndex.set(peer, resp.matchIndex);
      this.nextIndex.set(peer, resp.matchIndex + 1);
      return true;
    } else {
      // Log inconsistency: back off nextIndex and retry
      this.nextIndex.set(peer, Math.max(1, nextIdx - 1));
      return false;
    }
  }
}

This is simplified, production Raft implementations batch retries, use a pipeline of in-flight RPCs per peer, and handle the backoff more aggressively (using the conflict term optimization from the paper to skip entire terms of mismatched entries in one round-trip rather than one entry at a time).

Safety Guarantees

Raft provides two key safety properties:

Election Safety: At most one leader is elected per term. The quorum requirement and the “vote at most once per term” rule enforce this.

Log Matching: If two logs contain an entry with the same index and term, the logs are identical in all entries up through that index. This follows from the consistency check in AppendEntries and the fact that a leader never overwrites its own log.

Leader Completeness: If a log entry is committed in a given term, it will be present in the logs of all leaders for all higher-numbered terms. This follows from the log completeness constraint on voting: you cannot become leader unless your log is as up-to-date as a majority, and a majority will always overlap with the set of nodes that have the committed entry.

These three properties together mean that a committed entry will never be lost, regardless of how many leader failures occur.

Production Systems Using Consensus

etcd

etcd is the reference Raft implementation used by Kubernetes for all cluster state. It uses the etcd-io/raft library, which separates the Raft logic from the networking and storage layers. This means you can run Raft without etcd’s HTTP API if you want to embed it, and the core Raft logic is independently testable.

etcd uses a single Raft group for the entire key-value store. For most Kubernetes clusters this is fine because the bottleneck is not Raft throughput but the volume of state changes driven by controllers. Large clusters with many frequently reconciling resources do hit Raft limits; at that scale, etcd partitioning (multiple etcd clusters for different types of resources) is the intervention.

ZooKeeper

ZooKeeper uses ZAB (ZooKeeper Atomic Broadcast), not Raft or Paxos, but it solves the same problem with similar structure: a leader broadcasts ordered updates to followers, a majority must acknowledge before a write is committed, and leader election happens when the current leader fails. The differences are mostly in the details of recovery and leader epoch handling.

ZooKeeper predates Raft and is operationally more complex. It has its own ensemble sizing requirements (always use odd numbers), its own session semantics (watches, ephemeral nodes), and a specific data model (ZNodes) that bakes in assumptions about usage patterns. Most new projects that need consensus infrastructure reach for etcd over ZooKeeper because the operational surface is smaller.

CockroachDB

CockroachDB uses Raft but not a single global Raft group. Every range (a 64MB chunk of key-value data) has its own Raft group, and those groups are distributed across the cluster. A typical cluster has thousands of Raft groups running simultaneously. This architecture allows CockroachDB to scale horizontally in a way that a single-Raft-group system cannot.

The tradeoff: coordinating between ranges (for multi-range transactions) requires a two-phase commit protocol layered on top of Raft, because Raft only provides ordering within a single group. Cross-shard atomicity is a separate problem from within-shard ordering.

Algorithm Comparison

ConcernPaxosRaftZAB (ZooKeeper)
UnderstandabilityLow: many unspecified implementation detailsHigh: designed for understandabilityMedium: similar to Raft but with protocol-specific concepts
SafetyStrong (quorum required)Strong (quorum required)Strong (quorum required)
Liveness under partitionBlocked minorityBlocked minorityBlocked minority
Leader electionNot specified in basic PaxosFirst-class, term-basedFirst-class, epoch-based
Log gap handlingNot specifiedExplicit backoff + nextIndexExplicit epoch recovery
Multi-key transactionsRequires layeringRequires layeringRequires layering
Cluster reconfigurationNot specifiedJoint consensus or single-server changeDynamic reconfiguration in 3.5+
Production examplesSpanner, Chubby, Cassandra LWTetcd, CockroachDB, TiKVZooKeeper, Kafka (controller)
Embedded library availableLimitedYes (etcd-io/raft)No

When You Actually Need Consensus

Consensus is the right tool for a narrow set of problems. Before reaching for etcd or embedding Raft, it is worth being precise about what you actually need.

You need consensus when:

  • Exactly one node in a cluster must take an action at a time (leader election, primary selection).
  • You need to make a decision that multiple nodes must agree on before it takes effect (distributed transactions, configuration changes that must be atomic across the cluster).
  • You need a replicated state machine where all nodes apply the same operations in the same order.

You probably do not need consensus when:

  • You need distributed locking with lease semantics: a single etcd or ZooKeeper instance running separately already provides this. You do not need to run Raft in your application.
  • You need eventual consistency: if your use case tolerates replicas being temporarily inconsistent, a quorum-write/quorum-read pattern (as in Cassandra or DynamoDB) gives you tunable durability without the overhead of a consensus protocol.
  • You need to rate-limit or coordinate across nodes: a Redis instance or a simple database-backed counter handles this for the vast majority of throughput levels.
  • You are building something that runs on a single node or a single region with a managed database: the managed database’s replication is already handling consistency. Adding consensus infrastructure on top introduces operational complexity without benefit.

The failure modes to watch for: A common mistake is embedding a consensus library (or running an etcd cluster) for something that could be handled with a database-backed advisory lock and a heartbeat table. Consensus comes with real operational cost: cluster sizing (always odd numbers), upgrade coordination, disk I/O for the write-ahead log, and latency on every write that requires quorum acknowledgment. If your workload can tolerate occasional double-execution of a background job, a simpler lease-based approach with a database row and a TTL column will be more reliable in practice because it has fewer moving parts.

What Raft Does Not Solve

Raft guarantees ordering and durability within a single Raft group. It does not:

  • Provide serializability across multiple Raft groups (you need 2PC or a global ordering mechanism).
  • Handle Byzantine failures (a node that lies in its messages breaks Raft’s correctness). Raft assumes crash-stop failures only.
  • Guarantee bounded latency. A Raft leader under disk pressure can stall the entire cluster. Writes wait for fsync on the leader’s log before sending AppendEntries. A slow follower does not block commits (once the majority has acknowledged), but a slow leader does.
  • Replace a database. Raft orders writes and replicates them. What you do with those writes (the state machine), how you query state, and how you handle schema evolution are outside the protocol.

The Real Production Lesson

The hardest part of working with consensus in production is not implementing it. It is calibrating your mental model of what “committed” means end-to-end. A write acknowledged by Raft is durable in the Raft log. It is also sitting in the application’s in-memory state machine. If your application crashes after applying the entry but before persisting the result to a downstream system (a database, an external API), you have a consistency gap. Raft committed the operation. Your system’s observable state does not reflect it.

This is why systems like CockroachDB and TiKV treat Raft as the storage engine’s write-ahead log, not as a coordination layer sitting above storage. When Raft says a write is committed, it means the data is durably persisted, not just buffered in memory. If you are building something with an embedded Raft library, the state machine persistence question is yours to answer, and the answer matters.

Consensus algorithms eliminate one class of consistency problems. They do not eliminate the need to reason carefully about what your system’s invariants are and where they can be violated.

More in System Design

How Amazon Aurora Works Internally: The Log-Is-the-Database Architecture, Quorum Writes, and Storage-Compute Separation Behind Cloud-Native SQL
System Design ·

How Amazon Aurora Works Internally: The Log-Is-the-Database Architecture, Quorum Writes, and Storage-Compute Separation Behind Cloud-Native SQL

A deep dive into Aurora's storage-compute separation, the log-is-the-database design that pushes redo log processing to storage nodes, quorum-based replication across 6 copies in 3 AZs, protection group architecture, fast cloning, Serverless v2 scaling mechanics, and honest tradeoffs versus RDS, self-managed PostgreSQL, CockroachDB, and Cloud Spanner.

How CockroachDB Works Internally: Distributed SQL, Raft Consensus Per Range, and the Architecture Behind Serializable Transactions at Global Scale
System Design ·

How CockroachDB Works Internally: Distributed SQL, Raft Consensus Per Range, and the Architecture Behind Serializable Transactions at Global Scale

A deep dive into CockroachDB's internals: the 512MB range-based data model with automatic splitting, per-range Raft replication, MVCC timestamp ordering with hybrid logical clocks, DistSQL physical planning, serializable transactions with timestamp refreshes, closed timestamps for follower reads, online schema changes using multi-version schema descriptors, and production considerations for hotspots and write amplification.

How Container Runtimes Work Internally: Namespaces, Cgroups, and the OCI Stack From Docker Run to Process Isolation
System Design ·

How Container Runtimes Work Internally: Namespaces, Cgroups, and the OCI Stack From Docker Run to Process Isolation

Containers are not virtual machines and they are not magic. They are a thin composition of Linux kernel primitives: namespaces, cgroups, and a layered filesystem. This article traces exactly what happens between docker run and a running process, covering the OCI spec, containerd, runc, overlay filesystems, and container networking at the kernel level.

How YugabyteDB Works Internally: DocDB Storage, Tablet Splitting, and the Dual API Architecture That Scales PostgreSQL Horizontally
System Design ·

How YugabyteDB Works Internally: DocDB Storage, Tablet Splitting, and the Dual API Architecture That Scales PostgreSQL Horizontally

A deep dive into YugabyteDB internals covering the DocDB storage engine built on RocksDB LSM trees, per-tablet Raft consensus, YSQL and YCQL dual query layers, distributed MVCC with hybrid logical clocks, automatic tablet splitting and rebalancing, xCluster multi-region replication, and production considerations for hotspots, connection pooling, and schema design.