Consensus II: Raft and coordination services

Paxos is correct and efficient, but it has a reputation for being hard to understand, and harder still to turn into a complete system. In 2014 Diego Ongaro and John Ousterhout published Raft, a consensus algorithm designed with understandability as its primary goal. Raft provides the same guarantees as Multi-Paxos and similar performance, but it decomposes the problem into relatively independent parts, leader election, log replication and safety, and it specifies the practical details, such as membership changes and log compaction, that Paxos leaves to implementers. In a user study, students learned Raft more easily than Paxos. Raft is now the most widely implemented consensus algorithm: etcd, Consul, CockroachDB, TiKV, YugabyteDB, MongoDB's replication protocol, RabbitMQ quorum queues and Kafka's KRaft metadata quorum all use it or close derivatives. This lesson presents Raft in detail and then looks at the coordination services built on consensus.

Raft's model: a strong leader and a replicated log

A Raft cluster has a small odd number of servers, typically five, which tolerates two failures. At any time each server is a follower, a candidate or a leader. In normal operation there is exactly one leader and the others are followers. Clients send all commands to the leader. The leader appends each command to its log and replicates it to the followers; once a majority store the entry, it is committed, and every server applies committed entries, in log order, to its state machine. Raft's strong-leader design, in which log entries flow only from leader to followers, is the main simplification compared with Paxos.

Time is divided into terms, numbered with consecutive integers. Each term begins with an election, and at most one leader can be elected in a term. Terms act as a logical clock: every message carries the sender's term; a server that sees a larger term immediately adopts it and becomes a follower; and a server rejects messages from an older term. This is the fencing idea from lesson 4 built into the protocol, and it is how a deposed leader discovers that it has been replaced.

Leader election

Leaders send periodic heartbeats, empty AppendEntries messages, to every follower. A follower that hears nothing for an election timeout assumes the leader has failed. It increments its term, becomes a candidate, votes for itself and sends RequestVote messages to every other server. Each server votes for at most one candidate per term, first come first served, and persists its vote before replying. A candidate that collects votes from a majority becomes leader and starts sending heartbeats. If it instead hears from a leader with a term at least as large, it becomes a follower. If several candidates split the vote so that no one gets a majority, each times out and starts a new election with a higher term.

Split votes could repeat indefinitely, which is the FLP liveness problem from lesson 8 in another form. Raft's remedy is simple: election timeouts are chosen at random from an interval, such as 150 to 300 milliseconds. Usually one server times out well before the others, wins the election and sends heartbeats before anyone else becomes a candidate. The Raft paper's measurements showed elections completing in a few hundred milliseconds even under contention. The timeout must be much longer than the time to send a message, and much shorter than the typical time between failures, a condition the paper states as \(\text{broadcastTime} \ll \text{electionTimeout} \ll \text{MTBF}\).

Log replication

Each log entry contains a command, the term in which the leader received it, and an index, its position in the log. To replicate, the leader sends AppendEntries messages containing new entries, the index and term of the entry immediately preceding them, and the leader's commit index. A follower accepts the message only if its log contains an entry at that preceding index with that term; otherwise it rejects, and the leader retries with an earlier index until the logs match, then sends everything from that point. A follower that accepts deletes any conflicting entries after the matching point and appends the leader's entries. This simple consistency check maintains the Log Matching Property: if two logs contain an entry with the same index and term, the logs are identical in all entries up to that index.

An entry is committed once the leader has replicated it to a majority of servers, subject to one subtlety discussed below. The leader includes its commit index in later messages, so followers learn which entries are safe to apply. With a stable leader, committing a command takes one round trip from the leader to a majority, and many commands can be batched into each message.

Safety: the election restriction and the commit rule

Raft must guarantee that once an entry is committed, it is present in the log of every future leader; otherwise a new leader could overwrite a committed command and replicas would diverge. Two rules ensure this. The election restriction: a server grants its vote only if the candidate's log is at least as up to date as its own, comparing first the term of the last entry and then the log length. A committed entry is stored on a majority, and any candidate needs votes from a majority, so at least one voter has the entry, and that voter refuses any candidate lacking it. Hence every elected leader holds every committed entry.

The second rule concerns entries from earlier terms. A new leader may find entries from previous terms that are stored on a majority but were never known to be committed. Counting replicas alone is not enough to commit them, because the Raft paper shows a scenario in which such an entry could still be overwritten by a later leader. So Raft's commit rule says a leader commits only entries from its own current term by counting replicas; once such an entry is committed, all preceding entries are committed indirectly by the Log Matching Property. In practice a new leader commits a no-op entry at the start of its term to settle the earlier entries quickly.

The simulation below runs five Raft servers with randomized timeouts. After a leader is elected and commits two payments, a partition isolates the leader with one follower. The isolated leader keeps accepting a client's command but cannot commit it, because it cannot reach a majority. The other three elect a new leader in a higher term, which commits a new command. When the partition heals, the old leader sees the higher term, steps down, and its uncommitted entry is overwritten.

# A compact Raft simulation: election, log replication, a partition and healing.
import random
rnd = random.Random(450)
N = 5
class Node:
    def __init__(self, i):
        self.id, self.term, self.voted, self.log = i, 0, None, []      # log: list of (term, cmd)
        self.state, self.commit, self.votes = "follower", 0, set()
        self.reset_timer()
    def reset_timer(self): self.timer = rnd.randint(10, 20)          # randomized election timeout
nodes = [Node(i) for i in range(N)]
inbox, cut, events = [], set(), []          # cut: set of node pairs that cannot talk
def send(src, dst, msg):
    if (src, dst) not in cut and (dst, src) not in cut: inbox.append((dst, src, msg))
def last(n): return (n.log[-1][0] if n.log else 0, len(n.log))       # (last term, last index)

def step_down(n, term):
    if n.state == "leader": events.append(f"t={t:3d} node {n.id} sees term {term} and steps down")
    if term > n.term: n.term, n.voted = term, None
    n.state = "follower"

def handle(n, src, m):
    kind = m["kind"]
    if m["term"] > n.term: step_down(n, m["term"])
    if kind == "vote_req":
        up_to_date = (m["last_term"], m["last_idx"]) >= last(n)        # election restriction
        grant = m["term"] == n.term and n.voted in (None, src) and up_to_date
        if grant: n.voted = src; n.reset_timer()
        send(n.id, src, {"kind": "vote_rep", "term": n.term, "grant": grant})
    elif kind == "vote_rep" and n.state == "candidate" and m["term"] == n.term and m["grant"]:
        n.votes.add(src)
        if len(n.votes) > N // 2:
            n.state = "leader"; n.next = {j: len(n.log) for j in range(N)}; n.match = {j: 0 for j in range(N)}
            events.append(f"t={t:3d} node {n.id} becomes leader for term {n.term}")
    elif kind == "append":
        if m["term"] < n.term:
            send(n.id, src, {"kind": "append_rep", "term": n.term, "ok": False, "idx": 0}); return
        n.state = "follower"; n.reset_timer()
        p = m["prev_idx"]
        if p > len(n.log) or (p > 0 and n.log[p - 1][0] != m["prev_term"]):
            send(n.id, src, {"kind": "append_rep", "term": n.term, "ok": False, "idx": p}); return
        dropped = [c for (tm, c), new in zip(n.log[p:], m["entries"]) if (tm, c) != new]
        if dropped: events.append(f"t={t:3d} node {n.id} replaces uncommitted {dropped}")
        n.log = n.log[:p] + m["entries"]                              # overwrite any conflicting tail
        n.commit = max(n.commit, min(m["commit"], len(n.log)))
        send(n.id, src, {"kind": "append_rep", "term": n.term, "ok": True, "idx": len(n.log)})
    elif kind == "append_rep" and n.state == "leader" and m["term"] == n.term:
        if m["ok"]:
            n.match[src] = m["idx"]; n.next[src] = m["idx"]
            for k in range(len(n.log), n.commit, -1):                  # commit rule: majority and current term
                if n.log[k - 1][0] == n.term and 1 + sum(n.match[j] >= k for j in range(N) if j != n.id) > N // 2:
                    if k > n.commit: events.append(f"t={t:3d} leader {n.id} commits up to index {k}")
                    n.commit = k; break
        else:
            n.next[src] = max(0, n.next[src] - 1)                       # back up and retry

def tick(n):
    if n.state == "leader":
        n.match[n.id] = len(n.log)
        for j in range(N):
            if j != n.id:
                p = n.next[j]
                send(n.id, j, {"kind": "append", "term": n.term, "prev_idx": p,
                               "prev_term": n.log[p - 1][0] if p else 0,
                               "entries": n.log[p:], "commit": n.commit})
        return
    n.timer -= 1
    if n.timer <= 0:                                                   # start an election
        n.state, n.term, n.voted, n.votes = "candidate", n.term + 1, n.id, {n.id}
        n.reset_timer()
        lt, li = last(n)
        for j in range(N):
            if j != n.id: send(n.id, j, {"kind": "vote_req", "term": n.term, "last_term": lt, "last_idx": li})

def leader():
    ls = [n for n in nodes if n.state == "leader"]
    return max(ls, key=lambda n: n.term) if ls else None

for t in range(160):
    if t == 30: L = leader(); L.log += [(L.term, "pay 40"), (L.term, "pay 25")]
    if t == 50:                                   # partition: old leader + one follower vs the other three
        old = leader(); buddy = (old.id + 1) % N
        minority = {old.id, buddy}
        cut = {(a, b) for a in minority for b in range(N) if b not in minority}
        events.append(f"t={t:3d} partition: {sorted(minority)} | {sorted(set(range(N)) - minority)}")
        old.log.append((old.term, "refund 25 (old leader)"))
        events.append(f"t={t:3d} old leader {old.id} accepts 'refund 25' but cannot reach a majority")
    if t == 90:
        new = leader(); new.log.append((new.term, "pay 60 (new leader)"))
    if t == 120:
        cut = set(); events.append(f"t={t:3d} partition heals")
    batch, inbox = inbox, []
    for dst, src, m in batch: handle(nodes[dst], src, m)
    for n in nodes: tick(n)

print("\n".join(events))
for n in nodes:
    print(f"node {n.id} term {n.term} {n.state:9s} commit={n.commit} log={[c for _, c in n.log]}")

t= 12 node 0 becomes leader for term 1
t= 32 leader 0 commits up to index 2
t= 50 partition: [0, 1] | [2, 3, 4]
t= 50 old leader 0 accepts 'refund 25' but cannot reach a majority
t= 61 node 3 becomes leader for term 2
t= 92 leader 3 commits up to index 3
t=120 partition heals
t=121 node 0 sees term 2 and steps down
t=121 node 0 replaces uncommitted ['refund 25 (old leader)']
t=121 node 1 replaces uncommitted ['refund 25 (old leader)']
node 0 term 2 follower  commit=3 log=['pay 40', 'pay 25', 'pay 60 (new leader)']
node 1 term 2 follower  commit=3 log=['pay 40', 'pay 25', 'pay 60 (new leader)']
node 2 term 2 follower  commit=3 log=['pay 40', 'pay 25', 'pay 60 (new leader)']
node 3 term 2 leader    commit=3 log=['pay 40', 'pay 25', 'pay 60 (new leader)']
node 4 term 2 follower  commit=3 log=['pay 40', 'pay 25', 'pay 60 (new leader)']

Consider what the client of the old leader experienced: its command "refund 25" was accepted into a log but never committed, so the client never received a success reply and must retry, ideally with an idempotency key from lesson 2. Raft never acknowledges a command before it is committed, so no acknowledged command is lost.

Membership changes and log compaction

Clusters must sometimes add or replace servers. Switching directly from an old configuration to a new one is unsafe, because during the switch the old and new majorities might not overlap, allowing two leaders in one term. The original Raft paper used joint consensus, a transitional configuration in which decisions need majorities of both the old and new sets. Ongaro's dissertation later described a simpler method, changing membership one server at a time, since any majority of the old configuration and any majority of a configuration differing by one server always overlap; most implementations use this. etcd and other implementations also add new members as non-voting learners first, letting them catch up on the log before they count toward majorities.

Logs cannot grow forever. Raft servers periodically take a snapshot of their state machine, record the index and term of the last entry included, and discard the log up to that point. A follower that is too far behind receives the snapshot through an InstallSnapshot message instead of replaying entries it no longer has.

Reads in a consensus system

Writes go through the log, but what about reads? Reading from any follower can return stale data. Even reading from the leader is not automatically linearizable, because a leader that has been partitioned away may not yet know it has been replaced. Raft offers three options. The leader can append the read to the log like a write, which is correct but slow. With ReadIndex, the leader records its current commit index, confirms it is still leader by exchanging heartbeats with a majority, waits until it has applied entries up to that index, and then answers from its state machine. With leader leases, the leader assumes that no other leader can be elected for a period after a successful heartbeat round, shorter than the election timeout, and answers reads locally during that period; this is fast but depends on bounded clock drift, a timing assumption the other methods avoid. etcd, for example, offers linearizable reads by default and serializable, possibly stale, reads as an option for lower latency.

Raft in production: etcd

etcd is the clearest example of Raft packaged as a service. It stores a few gigabytes at most of key-value data, replicated across three or five members, and offers linearizable reads and writes, leases, watches that stream changes, and multi-key transactions. Every change increments a cluster-wide revision, which clients can use for optimistic concurrency and as a fencing token. Kubernetes stores all cluster state, every pod, service and configuration object, in etcd, which is why etcd's health is critical to every Kubernetes cluster and why its documentation recommends dedicated, fast disks: every write waits for a durable log append on a majority.

# Starting a three-member etcd cluster (one command per machine); member names illustrative
etcd --name ledger-1 \
  --initial-advertise-peer-urls https://10.0.1.11:2380 \
  --listen-peer-urls https://10.0.1.11:2380 \
  --listen-client-urls https://10.0.1.11:2379 \
  --advertise-client-urls https://10.0.1.11:2379 \
  --initial-cluster ledger-1=https://10.0.1.11:2380,ledger-2=https://10.0.1.12:2380,ledger-3=https://10.0.1.13:2380 \
  --initial-cluster-state new \
  --heartbeat-interval 100 \
  --election-timeout 1000

# A lease-based leader election from the command line: the key disappears if the
# holder stops renewing its lease, and every write carries a revision (fencing token).
etcdctl lease grant 10
etcdctl put --lease=<lease-id> /ledgerline/reconciler/leader "worker-7"
etcdctl get /ledgerline/reconciler/leader -w json   # shows mod_revision

Apache ZooKeeper, older than Raft, uses its own protocol, Zab (ZooKeeper Atomic Broadcast), which has the same shape: a leader, epochs that play the role of terms, a recovery phase that synchronizes followers with the new leader, and majority acknowledgment for each update. Zab guarantees that updates from one client are applied in order and that a new leader delivers every update committed by previous leaders before proposing new ones. ZooKeeper offers linearizable writes but, by default, reads served by any server, which may be stale; clients that need fresh reads call sync first. HashiCorp's Consul uses Raft for its catalog and key-value store.

Consensus for every partition

A single Raft group is limited by its leader's capacity, so large databases run many groups. CockroachDB, TiKV and YugabyteDB split data into ranges of tens to hundreds of megabytes, each replicated by its own Raft group, with leaders spread across machines. Spanner does the same with Paxos groups. This multi-Raft design scales consensus horizontally: a cluster of hundreds of machines may run hundreds of thousands of Raft groups. It combines with the partitioning techniques of lesson 10 and with the commit protocols of lesson 5 for transactions that span ranges.

Testing and verifying consensus implementations

Consensus protocols are short to describe and notoriously easy to implement wrongly. Raft's authors provided a TLA+ specification and a proof of safety; Amazon engineers described in 2015 how they used TLA+ to find subtle bugs in designs for DynamoDB, S3 and other services before writing code. Implementations are tested with fault injection, such as Jepsen's partitions and clock skew, and with deterministic simulation, in which the whole cluster runs in one process with a seeded scheduler so that any failing interleaving can be replayed exactly, a technique made famous by FoundationDB. The simulation in this lesson is a toy version of the same idea.

Worked examples