Distributed Consensus and State Machine Replication

2.1 Raft Protocol Mechanics

A replicated state machine works by agreeing on an ordered log of commands. Every replica applies the same commands in the same order, so every replica reaches the same state. Consensus is therefore a log problem rather than a state problem, and Raft is a protocol for maintaining that log.

Raft splits the problem into three parts that can be reasoned about separately: leader election, log replication, and safety. The split is the protocol's main contribution, because Paxos solved the same problem a decade earlier and resisted implementation.

Terms, roles, and elections

Time divides into terms, numbered with monotonically increasing integers. Each term has at most one leader. A term is a logical clock: every message carries a term number, and any node that sees a term higher than its own steps down to follower and adopts the higher term.

A node is a follower, a candidate, or a leader. Followers respond to messages and start nothing. A follower that receives no heartbeat within its election timeout becomes a candidate, increments the term, votes for itself, and requests votes. A candidate that collects a majority becomes leader.

Randomized election timeouts are the mechanism that makes elections terminate. Identical timeouts turn every follower into a candidate at the same instant, which splits the vote and repeats forever. Raft draws each timeout uniformly from a range, typically 150 to 300 milliseconds, so one node almost always times out first and wins before the others start.

Log replication and the matching property

The leader appends a client command to its own log and sends AppendEntries to every follower. The message carries the new entries plus the index and term of the entry immediately before them. A follower accepts the entries only when its log contains an entry at that index with that term. Otherwise it rejects, and the leader steps its index back and retries.

That one consistency check produces the Log Matching Property. If two logs hold an entry with the same index and the same term, then the entries are identical and every preceding entry is identical. The property holds by induction and it is what lets a leader repair a divergent follower by finding one agreement point and overwriting everything after it.

  index   1    2    3    4    5    6    7
  leader  1    1    1    4    4    5    5     (term of each entry)
  fol. A  1    1    1    4    4                 -> behind, just append
  fol. B  1    1    1    4    4    5    5       -> up to date
  fol. C  1    1    1    2    2    2            -> divergent tail

  For follower C the leader walks nextIndex back: 7, 6, 5, 4.
  At index 3 the term matches (1 == 1), so index 3 is the agreement point.
  The leader overwrites indexes 4 onward. Entries 4, 5, 6 in term 2 are
  discarded, and they were never committed, because a committed entry
  lives on a majority and this leader holds a majority.

An entry is committed once the leader replicates it to a majority. A committed entry is durable forever, because every future leader must hold it. That guarantee comes from the vote restriction rather than from the replication itself.

The two safety rules that implementations get wrong

The first rule is the election restriction. A voter grants its vote only if the candidate's log is at least as up to date as its own, where "up to date" compares the last entry's term first and the index second. Since any committed entry lives on a majority, and any winning candidate collects a majority, those two majorities intersect. The intersection holds the committed entry, so the candidate's log contains it or the voter refuses.

The second rule concerns committing entries from earlier terms. A new leader must not conclude that an old entry is committed merely because it now sits on a majority. The classic counterexample has an entry replicated to a majority, a leader change, that entry overwritten, and a later leader replicating it again. Raft forbids the shortcut: a leader commits an entry from a previous term only as a side effect of committing an entry from its own term. Most implementations append a no-operation entry at the start of each term for exactly this purpose.

Membership change

Changing the cluster from three nodes to five cannot happen in one step on every node at once, because during the transition some nodes use the old configuration and some use the new one. Two disjoint majorities then exist: two of the old three, and three of the new five. Two leaders get elected in the same term.

Raft offers two answers. Joint consensus uses a transitional configuration that requires majorities in both the old set and the new set for every decision. The cluster moves from old, to joint, to new, and at no point do two disjoint majorities exist. Single-server changes add or remove one node at a time, which guarantees overlap because a majority of N and a majority of N plus 1 always intersect.

Single-server change is simpler and it is what most implementations use. It has a subtlety: a new node starts with an empty log and cannot vote usefully until it catches up. Add the node as a non-voting learner first, let it replicate, and promote it only once its log is close to the leader's.

Snapshots and log compaction

The log grows forever. Compaction replaces a prefix with a snapshot of the state machine at that point, plus the index and term of the last included entry. A restarting node loads the snapshot and replays only the tail.

Two costs follow. Taking a snapshot of a large state machine is expensive, and doing it synchronously stalls the node. Implementations take it from a copy-on-write view or a fork, so the state machine keeps serving. And a follower that falls far behind needs a snapshot transfer rather than log entries, which the InstallSnapshot message provides in chunks.

The parameter that matters is the snapshot interval. Too frequent and the node spends its time snapshotting. Too rare and restart takes minutes and a lagging follower forces a large transfer. A common setting snapshots every 10,000 to 100,000 entries, tuned so that a restart replays under a few seconds.

2.2 Paxos Variants and Multi-Paxos

Paxos solves consensus for a single value. It was published in 1998, it is correct, and it is famous for being hard to translate into a system. Raft was designed as a reaction to that difficulty, and understanding Paxos explains what Raft chose to constrain.

Single-decree Paxos

Three roles exist: proposers, acceptors, and learners. One node usually plays all three. The protocol runs in two phases.

In phase one, a proposer chooses a proposal number higher than any it used before and sends Prepare(n) to the acceptors. An acceptor that has not promised a higher number replies with a promise not to accept anything below n, along with the highest-numbered proposal it has already accepted, if any.

In phase two, a proposer that collected promises from a majority sends Accept(n, v). The value v must be the value from the highest-numbered accepted proposal that any acceptor reported. Only if no acceptor reported one does the proposer get to choose its own value. A majority of accepts decides the value.

That constraint on v is the whole safety argument. Any two majorities intersect, so a proposer that completes phase one learns about every already-accepted value, and it must then propose that value.

  Acceptors: A1 A2 A3 A4 A5

  Round 1: proposer P1 gets Accept(1, "x") onto A1, A2, A3.
           "x" is now DECIDED, though nobody has learned it yet.

  Round 2: proposer P2 sends Prepare(2) and hears from A3, A4, A5.
           A3 reports: "I accepted (1, x)".
           A4, A5 report: nothing accepted.

  P2 must now propose "x", not its own value, because the majority
  {A3, A4, A5} intersects the majority {A1, A2, A3} at A3, and A3
  carries the evidence. Safety comes from that intersection, and
  from nowhere else.

The protocol has no liveness guarantee. Two proposers that keep raising their numbers can prevent each other from ever completing phase two, forever. That is not a bug in the protocol, it is the FLP result made concrete, and the practical answer is to elect a distinguished proposer.

Multi-Paxos

A replicated log needs consensus on a sequence of values, one per log slot. Running full Paxos per slot costs two round trips per entry, which is twice what is necessary.

Multi-Paxos observes that phase one is about the proposal number, not the slot. A proposer that runs phase one once, for all future slots, becomes a stable leader. Every subsequent entry costs only phase two, which is one round trip to a majority. Phase one runs again only after a leader change.

At that point Multi-Paxos and Raft are the same shape: a stable leader, one round trip per entry, and a recovery procedure after leader failure. The differences are in the details that Raft pins down and that Paxos leaves to the implementer, which is why two Multi-Paxos implementations often differ more from each other than either differs from Raft.

Flexible Paxos and the quorum insight

Flexible Paxos observes that the two phases do not both need a majority. They need quorums that intersect with each other. Phase one quorums must intersect phase two quorums, and phase two quorums need not intersect each other.

The consequence is practical. A system that changes leader rarely uses a small replication quorum and a large election quorum. With five nodes, replicate to two and require four votes to elect. Steady-state writes then wait for the second-fastest node instead of the third, which cuts tail latency, and leader election costs more, which is acceptable because it is rare.

EPaxos and leaderless execution

A leader is a bottleneck and a latency floor. Every write travels to the leader first, so a client far from the leader pays that distance twice.

Egalitarian Paxos removes the leader. Any replica commits a command directly. The replica attaches the set of commands it believes to be concurrent and interfering, and the protocol establishes a dependency graph rather than a total order. Commands that do not interfere commit in one round trip from the nearest replica. Commands that do interfere take a second round trip to agree on the ordering.

The gain is real for geographically distributed clusters with low conflict rates, because a client commits at its nearest replica. The costs are also real. Execution requires a topological sort of the dependency graph, and cycles need a deterministic tie-break. The implementation is substantially more complex than Raft, and conflict-heavy workloads degrade to worse than leader-based because every command takes the slow path.

Network partitions

A partition splits the cluster. Every protocol in this section responds the same way: the side with a majority continues and the side without it stops. That is the correct behavior and it is worth stating why.

A minority side that kept accepting writes produces a second history. Merging two histories is impossible in general, because both can hold committed decisions that contradict. Refusing writes on the minority side keeps the history single-valued, and the price is unavailability for the clients on that side.

Two failure modes deserve attention. A partitioned former leader that does not yet know it lost leadership will serve stale reads if it uses lease reads and its clock drifted. And a node that is partitioned from the leader but not from the voters will repeatedly time out, increment its term, and force elections, which disrupts an otherwise healthy cluster. The pre-vote extension fixes the second: a candidate first asks whether it can win, without incrementing its term. A node that nobody votes for then never disturbs the cluster.

2.3 Consistency Models and Time in Distributed Systems

A consistency model is a rule about which values a read can return. It is a contract between the store and the application, and choosing one is choosing what the application must handle.

Ordering without clocks

Physical clocks on different machines disagree. Network Time Protocol over the public internet holds a few milliseconds of error, and on a local network a few hundred microseconds. Any algorithm that compares timestamps from two machines without an error bound is wrong.

A Lamport timestamp is one counter per node. Increment it on every local event, send it with every message, and on receipt set the local counter to one more than the maximum of the local value and the received value. The result guarantees that if event A causes event B, then the timestamp of A is less than the timestamp of B. The converse does not hold: a smaller timestamp does not mean causation.

A vector clock holds one counter per node. Comparing two vectors gives three answers rather than two: A happened before B, B happened before A, or A and B are concurrent. That third answer is the value of the structure, because it identifies exactly the pairs that need conflict resolution.

  LAMPORT                          VECTOR
  A: 1 -- 2 ------ 3               A: [1,0,0] -- [2,0,0] ------ [3,1,0]
          \                                        \
           \ msg                                    \ msg
            v                                        v
  B: 1 ---- 3 ---- 4               B: [0,1,0] -- [2,2,0] -- [2,3,0]

  Lamport: event A@2 and event B@3 -- is A the cause of B? Unknown.
           The timestamps are ordered, and the order may be coincidence.

  Vector:  [2,0,0] vs [2,2,0]: every element of the first is <= the
           second, and one is strictly less, so the first happened before.
           [3,1,0] vs [2,3,0]: neither dominates, so they are CONCURRENT.
           A last-write-wins rule would silently discard one of them.

Vector clocks cost one entry per node, which grows with the cluster. Dynamo style systems bound the size by pruning the oldest entries, which trades a small risk of a false "happened before" for a fixed size.

Hybrid logical clocks and TrueTime

A hybrid logical clock combines both. The timestamp holds a physical component and a logical counter. The physical part stays within the clock error of real time, so the value is meaningful to a human and comparable against wall-clock queries. The logical part breaks ties and preserves causality exactly. CockroachDB and several other systems use hybrid logical clocks as their transaction timestamp.

Google TrueTime takes the opposite approach and makes uncertainty explicit. It returns an interval rather than an instant, with the guarantee that the true time lies inside it. Atomic clocks and GPS receivers in every datacenter keep the interval to a few milliseconds.

Spanner uses the interval in a specific way, and the mechanism is worth understanding because it is the only widely deployed system that buys external consistency with hardware. To commit, Spanner picks a timestamp at the upper bound of the current interval, then waits until the interval passes that timestamp before releasing locks. The wait, called commit wait, guarantees that the commit timestamp is in the past everywhere. Any later transaction therefore gets a strictly greater timestamp, which gives a global total order consistent with real time.

The cost is the wait itself, which is roughly the clock uncertainty on every read-write transaction. That is a few milliseconds, and it buys a property that no software-only design provides.

The model hierarchy

Linearizability is the strongest single-object model. Every operation appears to take effect at one instant between its invocation and its response, and that instant respects real time. Once a write returns, every later read returns that value or a newer one. It is what a compare-and-swap needs, and it is what "strong consistency" usually means.

Sequential consistency is weaker. All processes observe the same order of operations, and that order respects each process's own program order. It does not respect real time across processes, so a read that begins after a write returned can still observe the old value.

Causal consistency preserves only causally related order. If A causes B, then every observer sees A before B. Concurrent operations can appear in different orders to different observers. It is the strongest model that a system can provide while remaining available during a partition, which makes it the interesting ceiling for globally distributed stores.

Eventual consistency promises only that replicas converge if writes stop. On its own it is nearly useless to an application, and the session guarantees are what make it usable: read your writes, monotonic reads, monotonic writes, and writes follow reads.

Composition does not preserve consistency

Two linearizable stores composed into one system do not give a linearizable system. A service that writes to store A and then to store B has no atomicity across them, and a reader that sees B's write but not A's observes a state that never existed.

The consequence appears constantly in microservice architectures. Every service has a strongly consistent database, and the system as a whole provides no guarantee at all. Fixing it needs an explicit mechanism: a saga with compensations, an outbox with change data capture, or a single transaction boundary that covers both. Assuming that strong components compose into a strong system is the most common consistency error in practice.

2.4 Distributed Transactions and Two-Phase Commit

Consensus and atomic commit are different problems. Consensus agrees on one value that some node proposed, and any proposed value is acceptable. Atomic commit requires unanimity: every participant must agree to commit, and one veto aborts.

Two-phase commit mechanics

A coordinator drives the protocol. In the prepare phase it asks every participant whether it can commit. A participant that votes yes has made the change durable and holds its locks, and it has given up the right to abort unilaterally. In the commit phase the coordinator sends the decision, which is commit if every vote was yes and abort otherwise.

The protocol is correct. It is also blocking, and the blocking case is specific. If the coordinator fails after a participant voted yes and before that participant learns the decision, the participant cannot proceed. It cannot commit, because another participant perhaps voted no. It cannot abort, because the coordinator perhaps already told someone else to commit. It holds its locks and waits.

  coordinator          P1             P2             P3
       |  PREPARE ---->  |              |              |
       |  PREPARE ------------------->  |              |
       |  PREPARE ----------------------------------->  |
       |  <---- YES ---- |              |              |
       |  <---- YES -------------------  |             |
       |  <---- YES ----------------------------------  |
       |                 |              |              |
       |  X  coordinator crashes here                  |
       |                 |              |              |
       |            all three hold locks and wait.
       |            P1 cannot commit: maybe someone voted no.
       |            P1 cannot abort: maybe the decision was commit.
       |            The locks stay until the coordinator returns.

  Availability of the whole transaction is the PRODUCT of every
  participant's availability and the coordinator's.
  Five services at 99.9 percent give 99.5 percent together.

Three-phase commit adds a round to remove the blocking under a synchronous model, and it fails under network partition, which is the case that matters. The real fix is to make the coordinator fault tolerant by replicating its decision through consensus. Spanner does exactly that: each shard is a Paxos group, and the two-phase commit runs between groups with every step logged through Paxos, so the coordinator cannot disappear.

Sagas

A saga replaces the distributed transaction with a sequence of local transactions plus compensating actions. Each step commits locally and emits an event. If a later step fails, the saga runs compensating actions for the completed steps in reverse order.

Compensation is not rollback. A rollback restores the previous state as though nothing happened. A compensation performs a new, visible business action: a refund rather than an un-charge, a cancellation notice rather than a deleted booking. Design each compensating action with a domain expert, because it is a business decision rather than a technical one.

Sagas give up isolation, and the consequences are specific and nameable. A concurrent reader can observe a partial saga, which is a dirty read. Two sagas can interleave and produce a state that neither produces alone, which is a lost update. Three countermeasures exist: a semantic lock that marks a record as pending, a commutative update that is safe in any order, and a reread-and-check before the final step.

Orchestration puts the sequence in one component, which makes the flow visible and concentrates the coupling. Choreography spreads it across event handlers, which lowers coupling and leaves the flow existing only in the reader's head. Choose orchestration for flows longer than three steps or with real compensation, and choreography for simple notification chains.

The dual write problem and the outbox

A service that writes to its database and then publishes an event performs two writes with no transaction between them. A crash between them leaves the database updated and the event lost, or the reverse. Retrying does not help, because the failure happens between the two operations.

The transactional outbox fixes it. Write the event into an outbox table in the same database transaction as the state change. A relay reads the outbox and publishes, marking rows as sent. The database transaction makes the state change and the intent to publish atomic, and the relay gives at-least-once delivery to the broker.

-- One transaction covers the state change and the intent to publish.
BEGIN;

UPDATE orders
   SET status = 'SUBMITTED', version = version + 1
 WHERE order_id = $1 AND version = $2;      -- optimistic concurrency

INSERT INTO outbox (id, aggregate_id, event_type, payload, occurred_at)
VALUES (gen_random_uuid(), $1, 'OrderSubmitted', $3, now());

COMMIT;

-- The relay claims a batch without blocking other relays.
--   SELECT id, payload FROM outbox
--    WHERE published_at IS NULL
--    ORDER BY occurred_at
--    LIMIT 500
--    FOR UPDATE SKIP LOCKED;
-- Publish, then UPDATE outbox SET published_at = now() WHERE id = ANY($1);

Change data capture, which Section 4.1 covers, replaces the polling relay by reading the database write-ahead log directly. The outbox row appears in the log, a connector publishes it, and no polling query runs against the transactional store.

Deterministic execution

A different design removes the coordination entirely. If every replica executes the same transactions in the same order, and execution is deterministic, then no agreement on the outcome is needed. The replicas cannot disagree, because nothing in their execution can differ.

Calvin established the approach: a sequencing layer produces a global transaction order, and replicas execute it independently. VoltDB and TigerBeetle apply the same principle in a single-threaded, in-memory state machine. The result is very high throughput for the workloads that fit, because there is no lock manager, no two-phase commit, and no per-transaction coordination.

The constraints are strict. The transaction's full read and write set must be known before execution starts, which rules out a transaction whose second query depends on the first query's result. Execution must be deterministic, so no wall-clock reads, no random numbers, and no external calls. Where the workload fits those rules, the design outperforms everything else by a wide margin.

Sources and further reading

Diego Ongaro and John Ousterhout, "In Search of an Understandable Consensus Algorithm", and Ongaro's thesis for membership change, pre-vote, and read strategies. Leslie Lamport, "The Part-Time Parliament" and "Paxos Made Simple". Heidi Howard and others, "Flexible Paxos: Quorum Intersection Revisited". Iulian Moraru and others, "There Is More Consensus in Egalitarian Parliaments", SOSP 2013, for EPaxos. James Corbett and others, "Spanner", OSDI 2012, for TrueTime and commit wait. Alexander Thomson and others, "Calvin", SIGMOD 2012. Maurice Herlihy and Jeannette Wing, "Linearizability", ACM TOPLAS 1990. Chris Richardson, "Microservices Patterns", for sagas and the outbox. The Jepsen reports at jepsen.io for what these protocols do when implemented incorrectly.