Raft consensus architecture, in depth: deploying a Raft group, commit latency maths, placement, Multi-Raft and operations

Where consensus belongs in a system

Consensus gives you a replicated log that a majority of nodes agree on, and with it a linearizable state machine that stays available as long as a majority can talk to each other. It costs a majority round trip and durable writes on every update. Three architectural patterns use it:

• A consensus-backed metadata store. A small cluster, usually 3 or 5 nodes, holds configuration, membership, leases and locks for a larger system. etcd under Kubernetes is the standard example, and Kafka's KRaft controller quorum moved Kafka's own metadata onto a Raft log. Data volume is small and every update matters. See etcd in depth.
• Multi-Raft data storage. The keyspace is split into ranges, and each range is its own Raft group of three or five replicas spread over the cluster. CockroachDB, TiKV and YugabyteDB work this way. Consensus is on the data path, but no single group carries more than its share.
• A replicated log as the product. The log itself is the API, as in a replicated queue or a coordination journal.

The anti-pattern is putting bulk user data through one group. Every write then crosses the same leader and the same disks, and throughput stops at what one leader can replicate.

Where consensus belongs in a system

Consensus gives you a replicated log that a majority of nodes agree on, and with it a linearizable state machine that stays available as long as a majority can talk to each other. It costs a majority round trip and durable writes on every update. Three architectural patterns use it:

• A consensus-backed metadata store. A small cluster, usually 3 or 5 nodes, holds configuration, membership, leases and locks for a larger system. etcd under Kubernetes is the standard example, and Kafka's KRaft controller quorum moved Kafka's own metadata onto a Raft log. Data volume is small and every update matters. See etcd in depth.
• Multi-Raft data storage. The keyspace is split into ranges, and each range is its own Raft group of three or five replicas spread over the cluster. CockroachDB, TiKV and YugabyteDB work this way. Consensus is on the data path, but no single group carries more than its share.
• A replicated log as the product. The log itself is the API, as in a replicated queue or a coordination journal.

The anti-pattern is putting bulk user data through one group. Every write then crosses the same leader and the same disks, and throughput stops at what one leader can replicate.

The protocol in one paragraph

The protocol in one paragraph: time is divided into numbered terms with at most one leader per term. A follower that hears no heartbeat within its randomized election timeout becomes a candidate, increments the term and asks for votes. Each voter grants at most one vote per term, and only if the candidate's log is at least as up to date as the voter's own: a higher last-entry term wins, and with equal terms the longer log wins. A candidate with votes from a majority becomes leader. The leader appends client commands to its log and replicates them; an entry from the leader's current term is committed once a majority stores it, which also commits every earlier entry. Committed entries are applied to the state machine in log order. Any node that sees a higher term steps down. Because every two majorities overlap, a new leader always holds every committed entry.

Inside a node: from proposal to apply

Inside a node, a write passes through a pipeline, and its stages are where latency and bugs come from. The most widely used library, etcd's raft package, exposes it as a loop: the library hands the application a batch of work (a Ready) and the application must persist, send and apply it in a set order.

# The node loop around a Raft library (shape of etcd/raft's Ready handling)
while True:
    rd = raft.ready()                       # batched work since the last turn

    wal.append(rd.hard_state, rd.entries)   # term, vote, commit index + new entries
    wal.fsync()                             # MUST be durable before anyone relies on it
    if rd.snapshot:
        storage.apply_snapshot(rd.snapshot)

    transport.send(rd.messages)             # AppendEntries, votes, heartbeats
                                            # (leaders may send in parallel with their own fsync)

    for entry in rd.committed_entries:      # in log order, exactly once per index
        result = state_machine.apply(entry) # also persists applied_index atomically
        pending.complete(entry.index, result)   # wake the waiting client

    raft.advance()                          # tell the library this batch is done

Three rules matter here. The term, vote and log must be on disk before a node answers a vote or an append, or a crash can make it vote twice or forget an acknowledged entry. Applying must be deterministic and idempotent by index, because after a restart a node replays entries after its last persisted applied index. And batching is natural: one Ready may carry hundreds of client proposals, which share a single fsync and a single message per follower. That is where Raft's throughput comes from.

Predicting commit latency and throughput

A write commits when a majority, counting the leader, has it durable. The leader's own fsync and the replication to followers run concurrently, so commit latency is roughly the larger of the leader's fsync and the k-th fastest follower's round trip plus fsync, where k is the number of followers needed for a majority: 1 of 2 in a 3-node group, 2 of 4 in a 5-node group. Add the client's round trip to the leader, and the apply time.

def commit_latency_ms(leader_fsync, followers):
    """followers: list of (rtt_ms, fsync_ms) from the leader's point of view."""
    n = len(followers) + 1
    need = n // 2                           # followers needed besides the leader
    per_follower = sorted(rtt + fs for rtt, fs in followers)
    return max(leader_fsync, per_follower[need - 1])

# 3 nodes, 3 zones, NVMe: rtt about 1 ms, fsync about 0.5 ms
print(commit_latency_ms(0.5, [(1.0, 0.5), (1.2, 0.5)]))           # 1.5
# 5 nodes, 3 in a home region and 2 far away: fast commits, but losing the home region halts writes
print(commit_latency_ms(0.5, [(1, .5), (1.2, .5), (70, .5), (72, .5)]))   # 1.7
# same 5 nodes, but one nearby follower is down
print(commit_latency_ms(0.5, [(1, .5), (70, .5), (72, .5)] + [(1e9, 0)]))  # 70.5

The last line is the lesson: placement decides latency, and a single failure can move every write from local to cross-region speed. Disk matters as much as network. A cloud disk with a 10 ms fsync puts a 10 ms floor under every commit no matter how close the nodes are, and an fsync that stalls for hundreds of milliseconds delays heartbeats and can trigger elections. etcd's guidance is to keep the 99th percentile of WAL fsync under 10 ms. Throughput, by contrast, scales with batching and pipelining: a leader with several AppendEntries in flight per follower and hundreds of commands per batch can commit far more operations per second than one per round trip.

Sizing, placement and membership

Use odd numbers, spread voters over at least three failure domains, and never split a group evenly across two sites: with two regions, losing the one holding the majority stops writes, and there is no placement that survives either region failing. A third site, even one holding a single voter, fixes that.

Change membership one server at a time, which is how etcd and most production systems do it; the paper's joint consensus method allows arbitrary changes but is harder to implement. Add new nodes as learners, non-voting members that receive the log, and promote them to voters only once they have caught up, so the cluster never counts a node that cannot yet acknowledge writes. Never replace a failed voter by adding the new one before removing the old one in a 3-node group: for a moment there are four voters, majority three, and writes stall until the newcomer catches up; any further failure in that window stops the cluster.

Reads without losing linearizability

Reads are where deployments quietly lose linearizability. A leader that has been partitioned into a minority does not know it has been replaced and will happily serve stale reads from its local state. Safe options are a ReadIndex read, where the leader records its commit index, confirms leadership with a heartbeat round to a majority and answers once it has applied up to that index, or a lease read, which skips the round trip but trusts bounded clock drift. Followers can also serve linearizable reads by asking the leader for a read index. The full treatment, including why the leader must commit an entry in its own term first, is in Raft consensus in depth; the consistency model itself is explained in linearizability versus sequential consistency.

Multi-Raft: thousands of groups per node

A Multi-Raft node may host thousands of replicas of different groups. Run naively, each group would send its own heartbeats, its own fsyncs and its own timers, and an idle cluster would spend most of its CPU on overhead. Production systems share as much as possible:

• One write-ahead log or storage engine per node, so a single fsync covers entries for many groups.
• Coalesced heartbeats: one message per node pair carries liveness for all groups they share.
• Quiescence: groups with no traffic stop heartbeating until a write arrives.
• Leader balancing: leaders are spread evenly over nodes and moved near the clients that use them, because the leader does the most work and decides read latency.
• Split, merge and rebalance: ranges split when large or hot and merge when small; adding a replica elsewhere and removing one here is how data moves, using the same learner-then-promote steps.

Snapshots and log compaction

Without compaction the log grows forever. Periodically each node snapshots its state machine at an applied index and truncates the log before it. A follower that has fallen behind the truncation point cannot be caught up with entries any more; the leader ships it a snapshot instead. Snapshots of large state machines are expensive to send, so rate-limit them and keep enough log for a briefly offline node to catch up without one.

Worked example: elections caused by a disk

A team runs a 3-node etcd cluster for a large Kubernetes control plane, one node per zone, on general purpose cloud disks. After an upgrade adds many custom controllers, write rate rises and so do leader changes: several per hour, each causing a few seconds of API errors. Metrics tell the story. etcd_disk_wal_fsync_duration_seconds p99 has risen to 40 ms, with spikes above 500 ms when the disk's burst credits run out, longer than etcd's default 100 ms heartbeat interval and close to half the 1,000 ms election timeout. Followers miss heartbeats while the leader is stuck in fsync and call elections.

The fix is disk first: move the WAL to provisioned-IOPS volumes and fsync p99 drops to 3 ms. Elections stop. They then cut write volume by fixing a controller that rewrote unchanged objects. Moving to five nodes was considered and rejected: every commit would wait for three durable copies instead of two, to tolerate a second failure they did not need, while doing nothing about the disk.

Failure modes

• Slow fsync. Commit latency rises and heartbeats stall into elections. Watch WAL fsync duration and backend commit duration.
• Election storms. Timeouts too short for real network and disk latency. Enable PreVote and CheckQuorum, and alert on etcd_server_leader_changes_seen_total or its equivalent.
• Apply lag. The state machine applies slower than the log commits, so reads wait and memory grows. Track commit index minus applied index.
• Quorum loss. With a majority gone, writes stop by design. Recovery means forcing a new cluster from a survivor's data, which can lose acknowledged writes; document it and rehearse it.
• Oversized entries. Megabyte proposals block heartbeats behind them. Cap request size and store blobs elsewhere.

Trade-offs

For how majorities and overlapping quorums generalize beyond Raft, see Quorum systems.

What to do next

1. Write down which state in your system truly needs consensus, and keep bulk data out of a single group.
2. Place voters across three failure domains; confirm no two-site split can hold a majority.
3. Measure fsync p99 on the actual disks and compute expected commit latency with the formula above.
4. Check that election timeouts are many times your worst fsync and round trip, with PreVote enabled.
5. Confirm reads use ReadIndex or leases, not unconditional local reads on the leader.
6. Add new members as learners and promote them only once caught up.
7. Alert on leader changes, fsync latency, proposals pending and apply lag; rehearse quorum-loss recovery.