Consensus: Paxos & Raft

Consensus is the problem of getting a set of processes to agree on a single value. Formally, each process starts with an initial proposal (an input value), and the consensus protocol must satisfy three properties:

1. Agreement: No two correct processes decide different values. (Safety)
2. Validity: If all processes propose the same value v, then v must be the decided value. (Prevents trivially outputting a fixed constant.)
3. Termination: Every correct process eventually decides some value. (Liveness)

These three properties capture the intuitive meaning: the system must reach a decision (termination), the decision must be consistent (agreement), and the decision must be meaningful — it cannot be fabricated out of thin air (validity).

In the context of state machine replication, consensus is generalized: instead of agreeing on a single value, processes must agree on a sequence of values (a log of commands). The state machine replicated across nodes must apply the same commands in the same order, so every replica computes the same state. This is why the log abstraction is central to both Paxos and Raft.

A critical distinction: consensus is defined in a fault model. The standard model for these algorithms is crash-fault tolerance (CFT): processes may halt silently (crash), but they do not send incorrect messages (no Byzantine faults). A consensus algorithm tolerates f failures if it works correctly when at most f of n processes crash. As we will see, this requires n ≥ 2f + 1 (a strict majority must be alive).

Why a majority? Suppose you have 5 nodes and tolerate 2 crashes. Any decision must be made by a quorum — a set of nodes large enough that any two quorums overlap. If quorums are majorities (3 of 5), then any two quorums share at least one node. That shared node acts as a witness that carries information from one quorum to the next, preventing inconsistency.

# Consensus properties as executable assertions
# (checking a trace of a consensus run)

def check_consensus(decisions: dict[int, str], proposals: dict[int, str]) -> bool:
    """
    decisions: {process_id: decided_value} — only for processes that decided
    proposals: {process_id: proposed_value}
    """
    decided_values = list(decisions.values())

    # Agreement: all decided values must be the same
    assert len(set(decided_values)) <= 1, "Agreement violated!"

    # Validity: decided value must have been proposed by someone
    if decided_values:
        v = decided_values[0]
        assert v in proposals.values(), "Validity violated!"

    # Termination is a liveness property — cannot be checked on a
    # finite trace, but we can assert that all correct processes decided.
    # (assume all processes in proposals are 'correct' for this check)
    for pid in proposals:
        assert pid in decisions, f"Process {pid} never decided (termination violated)"

    return True

In 1985, Fischer, Lynch, and Paterson published a result so surprising it reshaped distributed systems theory: in a fully asynchronous system, it is impossible to solve consensus even if only one process may crash.

What 'fully asynchronous' means: Messages are eventually delivered (no permanent loss), but there is no upper bound on message delay. You cannot distinguish a crashed process from a very slow one.

The proof idea (bivalence argument):

Define a system configuration as bivalent if the future decision could still go either way (some execution leads to decision 0, another leads to decision 1). A configuration is univalent if the outcome is determined (0-valent: must decide 0; 1-valent: must decide 1).

1. Lemma 1 (Initial bivalence): Any protocol must have a bivalent initial configuration. If all processes start with 0, they must decide 0; if all start with 1, they decide 1. By validity, these are univalent. But if initial values differ, the initial configuration can be shown to be bivalent (by a connectivity argument — changing one process's input from 0 to 1 one at a time, there must be a boundary where bivalence holds).

2. Lemma 2 (Staying bivalent): From any bivalent configuration, there exists a bivalent successor — i.e., any protocol can always be kept bivalent by an adversarial scheduler. The proof shows that if applying any particular message m always resolved bivalence, you could construct a contradiction involving a crashed process and message reordering.

Putting these together: the adversary can keep any protocol in a bivalent state forever by carefully delaying messages. Since the protocol can never be forced to decide, termination is violated.

What this means in practice: Real systems break the impossibility by weakening the model:

• Partially synchronous systems: assume messages are eventually delivered within some unknown bound Δ. This is the model Paxos and Raft operate in.
• Randomized algorithms: use coin flips to escape adversarial scheduling (e.g., Ben-Or's algorithm).
• Failure detectors: add unreliable oracles that hint about crashed processes.

FLP does NOT say consensus is impossible in practice. It says: if your system has no timing assumptions at all (pure asynchrony), consensus cannot be solved deterministically. Every real consensus algorithm implicitly assumes partial synchrony — it needs timeouts to detect failures, and those timeouts only make sense in a world with eventual timing bounds.

Paxos solves a restricted version of consensus: agreeing on a single value (one decree). Multi-Paxos extends this to a log, but understanding single-decree Paxos is essential.

Paxos has three roles:

• Proposer: wants to get a value accepted.
• Acceptor: votes on proposals.
• Learner: learns the decided value.

(In practice, every node plays all three roles.)

The algorithm proceeds in two phases:

Phase 1: Prepare / Promise

A proposer chooses a unique, monotonically increasing proposal number n (a ballot number). It sends Prepare(n) to a quorum (majority) of acceptors.

Each acceptor responds with a Promise(n, accepted_n, accepted_v) if n is larger than any prepare it has previously responded to. The promise means: I will not respond to any prepare or accept with a number less than n. The acceptor also reports the highest-numbered proposal it has already accepted (if any).

Phase 2: Accept / Accepted

If the proposer receives promises from a quorum, it proceeds. It chooses a value:

• If any acceptor reported an accepted value, the proposer must use the accepted value with the highest proposal number. (This is the critical safety invariant.)
• Otherwise, the proposer is free to propose any value (including its own original proposal).

The proposer sends Accept(n, v) to a quorum of acceptors. Each acceptor accepts unless it has since promised a higher number. When an acceptor accepts, it notifies learners.

Why does this work?

The key insight is the chose-the-highest-accepted-value rule. Suppose value v was accepted by a quorum Q1 in round n. Any future proposer in round n' > n must get promises from a quorum Q2. Since Q1 and Q2 overlap (both are majorities), at least one node in Q2 accepted v in round n. That node reports it in its Promise response. The new proposer therefore picks v (it has the highest accepted ballot number), preserving the already-chosen value.

Paxos limitations:

• It only chooses a single value.
• The protocol can livelock: two proposers can keep bumping each other's ballot numbers forever.
• Lamport's paper was famously hard to understand; the 1998 published version is called 'Paxos Made Simple' but is still considered difficult.
• Engineering a full replicated log on top of Paxos (Multi-Paxos) requires solving many problems the paper leaves unspecified (leader election, log gaps, membership changes).

# Paxos single-decree — minimal pseudocode

# ---- Acceptor state ----
class Acceptor:
    def __init__(self):
        self.promised_n = -1      # highest prepare we've responded to
        self.accepted_n = -1      # ballot number of accepted proposal
        self.accepted_v = None    # value of accepted proposal

    def on_prepare(self, n):
        if n > self.promised_n:
            self.promised_n = n
            return ('promise', n, self.accepted_n, self.accepted_v)
        else:
            return ('nack', self.promised_n)

    def on_accept(self, n, v):
        if n >= self.promised_n:
            self.promised_n = n
            self.accepted_n = n
            self.accepted_v = v
            return ('accepted', n, v)
        else:
            return ('nack', self.promised_n)

# ---- Proposer logic ----
def run_proposer(my_value, acceptors, quorum_size):
    n = generate_unique_ballot()   # globally unique, monotone

    # Phase 1
    promises = [a.on_prepare(n) for a in acceptors]
    promises = [p for p in promises if p[0] == 'promise']
    if len(promises) < quorum_size:
        return None  # retry with higher n

    # If any acceptor already accepted something, use that value
    best = max(promises, key=lambda p: p[2])  # highest accepted_n
    v = best[3] if best[2] >= 0 else my_value

    # Phase 2
    accepted = [a.on_accept(n, v) for a in acceptors]
    accepted = [a for a in accepted if a[0] == 'accepted']
    if len(accepted) >= quorum_size:
        return v   # consensus reached on v
    return None    # retry

Single-decree Paxos agrees on one value. Real systems need to agree on a sequence of commands — a replicated log. Multi-Paxos extends single-decree Paxos by running a separate Paxos instance for each log slot.

The key optimization: Running a full two-phase Paxos for every single log entry would be expensive. Multi-Paxos amortizes Phase 1 across many slots:

1. A leader is elected by running Phase 1 on all future slots at once.
2. The leader can then run only Phase 2 (Accept) for each new log entry as long as it remains leader.
3. This reduces the common-case write to a single round-trip (just the Accept phase).

State Machine Replication (SMR): The log is the mechanism; the state machine is the point. Each replica maintains:

• A replicated log of commands (entries like SET x 5, DELETE y).
• A state machine that applies committed log entries in order.

Because all replicas apply the same commands in the same order, they reach the same state. This is the fundamental theorem of SMR: a deterministic state machine, fed the same input sequence, produces the same output sequence. Clients interact with the leader, which appends to the log. Once an entry is committed (accepted by a quorum), it is safe to apply to the state machine and respond to the client.

The tricky bits in Multi-Paxos (which Lamport's paper leaves largely unspecified):

• Log gaps: If a proposer dies after Phase 1 but before Phase 2, a slot may have no accepted value. The new leader must run Phase 1 on that slot and either fill it with a no-op or drive the previously proposed value to completion.
• Leader uniqueness: There can briefly be two nodes that both believe they are leader (during an election). The ballot number mechanism ensures only one can successfully commit — old leaders' Phase 2 Accept messages will be rejected by acceptors that have promised higher ballots.
• Membership changes: Adding or removing nodes requires careful protocol extensions (usually a separate consensus round to agree on the new configuration).

These complexities motivated the design of Raft.

# State machine replication — the core abstraction

class ReplicatedStateMachine:
    def __init__(self):
        self.log = []            # [(term, command), ...], committed prefix
        self.state = {}          # the actual key-value state
        self.commit_index = -1   # highest committed log index
        self.last_applied = -1   # highest applied log index

    def apply_committed(self):
        """Apply all committed-but-not-yet-applied log entries."""
        while self.last_applied < self.commit_index:
            self.last_applied += 1
            cmd = self.log[self.last_applied][1]
            self._apply(cmd)

    def _apply(self, cmd):
        op, key, *rest = cmd
        if op == 'SET':
            self.state[key] = rest[0]
        elif op == 'DELETE':
            self.state.pop(key, None)
        elif op == 'NOOP':
            pass  # no-op used to commit previous entries on leader election

    def advance_commit(self, new_commit_index):
        """Called when consensus confirms entries up to new_commit_index."""
        self.commit_index = max(self.commit_index, new_commit_index)
        self.apply_committed()

Raft was created by Diego Ongaro and John Ousterhout (Stanford, 2014) with an explicit goal: be as understandable as possible while providing equivalent safety guarantees to Multi-Paxos. Their user study showed Raft was significantly easier for students to understand than Paxos.

Raft decomposes the consensus problem into three relatively independent sub-problems:

1. Leader election — choosing one leader at a time.
2. Log replication — the leader accepts client commands and replicates them to followers.
3. Safety — ensuring that if any server has applied a log entry at index i, no other server will apply a different entry at index i.

Core concepts:

Terms: Raft divides time into terms, numbered with consecutive integers. Each term begins with an election. If a candidate wins, it serves as leader for the rest of the term. Terms act as logical clocks — they help detect stale leaders.

Roles: Each node is in one of three states: follower, candidate, or leader. All nodes start as followers.

Heartbeats: Leaders send periodic AppendEntries RPCs (which can carry log entries or be empty heartbeats) to followers. If a follower receives no communication within an election timeout (randomized, typically 150–300 ms), it assumes the leader has failed and starts an election.

The key simplification over Paxos: Raft enforces a strong leader property — log entries only flow from leader to followers, never in the reverse direction. This single constraint dramatically simplifies the protocol. There is no need for complex quorum-intersection reasoning at the follower level; followers simply accept what the leader sends (if the leader has the right term and the log prefix matches).

# Raft node state — the data model
from enum import Enum
from dataclasses import dataclass, field
from typing import Optional

class Role(Enum):
    FOLLOWER = 'follower'
    CANDIDATE = 'candidate'
    LEADER = 'leader'

@dataclass
class LogEntry:
    term: int      # term when entry was received by leader
    command: object

@dataclass
class RaftNode:
    id: int
    peers: list[int]

    # Persistent state (must survive crashes — write to stable storage before responding to RPCs)
    current_term: int = 0
    voted_for: Optional[int] = None   # candidateId voted for in current term
    log: list[LogEntry] = field(default_factory=list)  # log[0] = first entry

    # Volatile state on all servers
    commit_index: int = -1   # highest log index known to be committed
    last_applied: int = -1   # highest log index applied to state machine

    # Volatile state on leaders (reinitialized after each election)
    next_index: dict = field(default_factory=dict)   # {peer_id: next log index to send}
    match_index: dict = field(default_factory=dict)  # {peer_id: highest log index known replicated}

    # Runtime state
    role: Role = Role.FOLLOWER
    leader_id: Optional[int] = None
    votes_received: int = 0

Leader election in Raft works as follows:

Starting an election: When a follower's election timeout fires (no heartbeat received), it:

1. Increments its current_term.
2. Transitions to candidate.
3. Votes for itself.
4. Sends RequestVote RPCs to all other nodes in parallel.

Granting a vote: A server grants its vote if:

1. The candidate's term is at least as large as the voter's current term.
2. The voter has not yet voted in this term (or already voted for this candidate).
3. The candidate's log is at least as up-to-date as the voter's log (the election restriction).

Log up-to-dateness is defined by comparing the last log entry: the candidate's log is more up-to-date if its last entry has a higher term; if terms are equal, the longer log wins.

Winning the election: A candidate wins if it receives votes from a majority. It immediately becomes leader and sends heartbeats to establish authority.

Randomized timeouts — the key to liveness: If two followers time out simultaneously, they split the vote and nobody wins. Raft resolves this with randomized election timeouts (each node picks independently from a range like 150–300 ms). The node that times out first is likely to win before others start their own elections. This is simple and works well in practice.

Term numbers prevent old leaders: If a partitioned old leader rejoins, it will see a higher term in any RPC it receives. On seeing a higher term, any node immediately reverts to follower. Old leaders cannot commit new entries because their Accept messages carry a stale term and will be rejected.

The election restriction ensures safety: This is the subtle but crucial invariant. When a new leader is elected, it must have all committed entries in its log. Why? Because committed entries were accepted by a quorum. Any candidate must get votes from a quorum. The two quorums overlap. The overlapping voter has the committed entry. If the candidate's log were missing that entry, the voter's log would be more up-to-date, and the voter would deny the vote. Therefore, a candidate without committed entries cannot win.

# Raft RequestVote RPC handler

def on_request_vote(node, args):
    """
    args: {term, candidate_id, last_log_index, last_log_term}
    returns: {term, vote_granted}
    """
    # Rule: if we see a higher term, revert to follower
    if args['term'] > node.current_term:
        node.current_term = args['term']
        node.voted_for = None
        node.role = Role.FOLLOWER

    if args['term'] < node.current_term:
        return {'term': node.current_term, 'vote_granted': False}

    # Check if we can grant the vote
    already_voted = (node.voted_for is None or
                     node.voted_for == args['candidate_id'])

    # Log up-to-dateness check
    my_last_term = node.log[-1].term if node.log else 0
    my_last_idx = len(node.log) - 1
    candidate_last_term = args['last_log_term']
    candidate_last_idx = args['last_log_index']

    candidate_up_to_date = (
        candidate_last_term > my_last_term or
        (candidate_last_term == my_last_term and
         candidate_last_idx >= my_last_idx)
    )

    if already_voted and candidate_up_to_date:
        node.voted_for = args['candidate_id']
        # persist current_term and voted_for to stable storage here!
        return {'term': node.current_term, 'vote_granted': True}

    return {'term': node.current_term, 'vote_granted': False}

Once a leader is elected, it handles client requests:

1. Append to local log: The leader appends the command as a new log entry with the current term.
2. Replicate via AppendEntries: The leader sends AppendEntries RPCs to all followers in parallel. Each RPC includes:
   • term and leader_id
   • prev_log_index and prev_log_term: the index and term of the log entry immediately preceding the new ones
   • entries[]: the new log entries (empty for heartbeats)
   • leader_commit: the leader's current commit_index
3. Commit: Once the leader hears back from a majority (including itself), it advances its commit_index and applies the entry to its state machine. It also notifies followers of the new commit_index in subsequent AppendEntries calls.
4. Follower consistency check: A follower only appends new entries if the (prev_log_index, prev_log_term) pair matches its own log. This is the Log Matching Property: if two entries in different logs have the same index and term, then the logs are identical in all entries up through that index.

Handling inconsistencies (log repair): When a new leader is elected, some followers may have stale logs (missing entries from old leaders that failed before committing). The new leader must bring them up to date. For each follower, the leader maintains next_index (the next entry to send). If an AppendEntries consistency check fails, the leader decrements next_index and retries. The follower's log is brought into sync by overwriting any conflicting entries.

Committing entries from previous terms: This is a subtle point. A new leader cannot directly commit log entries from a previous term by counting replicas — it must first commit an entry from the current term. This is because an entry with an old term number can appear to be replicated on a majority, yet still be safely overwritten by a future leader. The new leader appends a no-op entry in the current term to commit all prior entries as a side effect. This is the source of one of the most famous Raft figures (Figure 8 in the Raft paper).

Write linearizability: Because all writes go through the leader, and the leader commits before responding to clients, Raft provides linearizable (strongly consistent) reads and writes — the gold standard for distributed systems correctness.

# Raft AppendEntries RPC handler (follower side)

def on_append_entries(node, args):
    """
    args: {term, leader_id, prev_log_index, prev_log_term,
           entries, leader_commit}
    returns: {term, success}
    """
    if args['term'] > node.current_term:
        node.current_term = args['term']
        node.voted_for = None
        node.role = Role.FOLLOWER

    # Reset election timeout — we heard from a valid leader
    node.reset_election_timer()
    node.leader_id = args['leader_id']

    if args['term'] < node.current_term:
        return {'term': node.current_term, 'success': False}

    prev_idx = args['prev_log_index']
    prev_term = args['prev_log_term']

    # Log consistency check
    if prev_idx >= 0:
        if prev_idx >= len(node.log):
            return {'term': node.current_term, 'success': False}  # missing entry
        if node.log[prev_idx].term != prev_term:
            # Conflict: delete this entry and all that follow
            node.log = node.log[:prev_idx]
            return {'term': node.current_term, 'success': False}

    # Append any new entries, overwriting conflicting ones
    insert_idx = prev_idx + 1
    for i, entry in enumerate(args['entries']):
        pos = insert_idx + i
        if pos < len(node.log):
            if node.log[pos].term != entry.term:
                node.log = node.log[:pos]  # truncate conflicting suffix
                node.log.append(entry)
        else:
            node.log.append(entry)

    # Advance commit index
    if args['leader_commit'] > node.commit_index:
        node.commit_index = min(args['leader_commit'],
                                len(node.log) - 1)
        node.apply_committed()

    # persist log and current_term to stable storage here!
    return {'term': node.current_term, 'success': True}

Raft's correctness is captured by a set of invariants that the protocol maintains at all times:

1. Election Safety: At most one leader can be elected per term. (Proof: a candidate needs a majority of votes; since each server votes at most once per term, two candidates cannot both get majorities.)

2. Leader Append-Only: A leader never overwrites or deletes entries in its log — it only appends. This makes the leader the single source of truth.

3. Log Matching: If two logs contain an entry with the same index and term, then the logs are identical in all entries up through that index. (Proof by induction: the base case is that leaders only create one entry per index per term. The inductive step uses the AppendEntries consistency check.)

4. Leader Completeness: If a log entry is committed in a given term, that entry will be present in the logs of all leaders for all higher-numbered terms. (Proof: uses the election restriction — a leader must have all committed entries or it could not win the election.)

5. State Machine Safety: If a server has applied a log entry at a given index to its state machine, no other server will ever apply a different log entry for that index. (Follows directly from Log Matching + Leader Completeness.)

The big picture: These invariants together guarantee that all replicas eventually apply the same log entries in the same order. Since the state machine is deterministic, all replicas reach the same state. This is the definition of correct state machine replication.

Cluster membership changes: Changing the cluster configuration (adding/removing nodes) is dangerous — a naive approach can lead to two disjoint majorities. Raft handles this with joint consensus: a transitional period where decisions require a majority from both the old and new configurations, ensuring no two leaders can exist simultaneously during the transition. Simpler implementations use a single-server change protocol that only adds/removes one node at a time, which is safe because you cannot split a majority by changing one node.

Log compaction (snapshots): Logs grow without bound. Raft uses snapshots: the state machine periodically writes its current state to stable storage, and the log entries up to that point are discarded. When a follower is far behind, the leader can send an InstallSnapshot RPC instead of replaying all log entries.

# Checking the Log Matching invariant between two replicas
# (useful for testing a Raft implementation)

def check_log_matching(log_a: list, log_b: list) -> bool:
    """
    If two log entries at the same index have the same term,
    all preceding entries must be identical.
    log format: [(term, command), ...]
    """
    min_len = min(len(log_a), len(log_b))
    for i in range(min_len):
        if log_a[i][0] == log_b[i][0]:  # same term at index i
            # All entries 0..i must be identical
            assert log_a[:i+1] == log_b[:i+1], (
                f"Log Matching violated at index {i}: "
                f"same term {log_a[i][0]} but logs differ before index {i}"
            )
    return True

Both Paxos and Raft rely on quorums — subsets of nodes that overlap in a useful way. Understanding the mathematics here deepens your intuition for why these protocols work.

A quorum system for n nodes is a collection of subsets (quorums) such that any two quorums intersect. The simplest quorum system is majority quorums: every subset of size ⌊n/2⌋ + 1.

Why majority? With n = 2f + 1 nodes, any two majorities have at least one node in common. This overlapping node acts as a witness that carries information between protocol phases — it tells the new proposer what the old proposer already committed.

Availability vs. consistency:

• To tolerate f crash failures, you need n = 2f + 1 nodes. You can lose f nodes and still have a majority of f + 1.
• This is a hard lower bound: with n = 2f nodes, a perfectly symmetric partition (f on each side) is indistinguishable from f failures, and neither partition can form a majority. One side must sacrifice availability.

The CAP theorem (Brewer, 2000; proved by Gilbert & Lynch, 2002): In the presence of a network partition, a distributed system can guarantee either Consistency (all reads see the latest write) or Availability (all requests receive a response), but not both. Paxos and Raft choose consistency: if the partition isolates fewer than a quorum, the system blocks (sacrifices availability) rather than risk inconsistency.

Flexible quorums: Modern systems (like CockroachDB) use hierarchical quorums that account for data center geography — a write must be acknowledged by a majority of data centers, and within each DC, a majority of nodes. This reduces cross-DC latency for reads without sacrificing safety.

Quorum read optimization: In Raft, all reads through the leader are trivially consistent. But having every read go through the leader is a bottleneck. Raft can serve reads from followers with lease-based reads (a leader holds a lease valid for some time window, during which no other leader can exist) or linearizable reads (the leader commits a no-op to confirm its leadership before responding).