Distributed Systems Foundations and Consensus Protocols

What the network takes away

A local function call either returns or throws. A remote call has a third outcome: no answer. The caller cannot distinguish a crashed server, a dropped request, a slow server, a dropped response, or a network partition. Every one of those looks the same from the outside, and the difference between them changes what the correct action is.

That single fact produces every hard problem in this chapter. The request perhaps ran already. A retry then runs it twice, and no retry then loses it. The only escape is to make the operation idempotent, which means that running it twice has the same effect as running it once. Idempotence is a design property that the operation must be built to have.

Three other properties disappear with the network. There is no global clock, so "before" is not well defined across machines. There is no shared memory, so state exists only where a message put it. There is no reliable failure detector, which the next section makes precise.

Every consensus protocol in this chapter costs at least one round trip to a majority. Put the numbers from Table 8.1 into that sentence and the geography of a deployment becomes a latency budget. A three-region synchronous quorum across the United States and Europe commits in at least 70 milliseconds, and no protocol improvement changes that, because the limit is the speed of light in glass.

Failure models and synchrony

A failure model states what a faulty component can do. Crash-stop means that a node stops and never returns. Crash-recovery means that it stops and comes back, possibly having lost volatile state. Omission means that it drops messages. Byzantine means that it can do anything at all, including sending different answers to different peers.

Almost every system inside one administrative domain assumes crash-recovery. Byzantine fault tolerance costs roughly three times the replicas and several times the messages. It earns that cost only where the participants do not trust each other, which means blockchains and some regulated multi-party settlements.

A synchrony model states what you know about timing. A synchronous system has known bounds on message delay and processing time. An asynchronous system has no bounds at all. A partially synchronous system behaves asynchronously for an unknown period and then becomes synchronous, which describes the real internet well enough to build on.

The FLP result, from Fischer, Lynch, and Paterson in 1985, proves that no deterministic algorithm solves consensus in an asynchronous system when even one node can crash. The proof is not a practical obstacle. It is a precise statement of what the obstacle is. You cannot guarantee both an answer and correctness when you cannot tell a slow node from a dead one.

Real systems escape FLP in three ways, and every production protocol uses at least one. They assume partial synchrony and use timeouts, which gives safety always and liveness when the network behaves. They use randomization, so that termination happens with probability one. They use an unreliable failure detector as an oracle, which moves the assumption into a named component.

Raft, Paxos, Zab, and Viewstamped Replication all take the first path. They never produce a wrong answer, whatever the network does. They stop making progress when the network stays bad, and they resume when it recovers. That trade is almost always the right one, because a wrong answer in a bank ledger costs more than a pause.

CAP, PACELC, and the common misreading

The CAP theorem states that a system that keeps replicas cannot have both linearizability and total availability when a network partition splits it. That is the whole claim, and it is narrower than its reputation.

Three corrections matter. First, the choice applies only during a partition. A system with no partition can have both. Second, "availability" in CAP means that every non-failed node answers every request, which is stronger than any service level objective. Third, consistency in CAP means linearizability specifically, not "consistency" in the ACID sense and not any weaker model.

"Pick two of three" is therefore wrong. Partitions are not a choice, they are an event. The real choice is what to do when one happens: refuse requests on the minority side, or answer them and reconcile later.

PACELC, from Daniel Abadi, adds the part that governs the other 99.99 percent of the time. If there is a partition, choose between availability and consistency. Else, choose between latency and consistency. The second clause is the one that decides most architectures, because synchronous replication to a distant replica costs latency on every request whether or not anything ever fails.

The Cassandra and DynamoDB rows show that the classification is a configuration, not a product property. The same store gives different answers depending on the consistency level that the client asks for, and a team that picks the level per call site gets the trade right per use case.

The consistency hierarchy

Consistency models form a hierarchy. Each one is a rule about which values a read can return.

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 "strong consistency" usually means, and it is what a compare-and-swap needs.

Sequential consistency is weaker. All processes see 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 starts after a write returns can still see the old value.

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

Eventual consistency promises only that replicas converge if writes stop. It says nothing about what a read returns in the meantime, which is why it needs the session guarantees below to be usable.

Session guarantees fix the worst user-visible problems without full strength. Read your writes means that a client sees its own writes. Monotonic reads means that a client never sees time go backward. Monotonic writes means that one client's writes apply in order. Writes follow reads means that a write that follows a read applies after the value that the read saw. Most "eventually consistent" systems that feel correct are providing these four.

  LINEARIZABLE            single global order, respects real time
     |                    cost: consensus, 1 majority round trip per op
     |                    use: locks, leader election, account balance
     v
  SEQUENTIAL              single global order, ignores real time
     |                    cost: total order broadcast
     v
  CAUSAL                  causally related ops ordered, rest concurrent
     |                    cost: dependency tracking (vector clocks)
     |                    use: comments, social feeds, collaborative docs
     |                    the strongest model available under partition
     v
  SESSION GUARANTEES      read-your-writes, monotonic reads/writes
     |                    cost: sticky routing or a version token
     v
  EVENTUAL                replicas converge if writes stop, no order
                          cost: almost nothing
                          use: caches, counters, presence

Time and ordering

Physical clocks disagree. NTP over the public internet holds a few milliseconds of error, NTP on a local network holds hundreds of microseconds, and PTP with hardware timestamping holds under a microsecond. Cloud providers now run PTP-backed time services that quote error bounds in the tens of microseconds. None of these is exact, and an algorithm that compares timestamps from two machines without an error bound is wrong.

Last-write-wins conflict resolution based on wall clock timestamps silently loses data for exactly this reason. Two writes arrive within the clock skew window, the one with the larger timestamp wins, and the larger timestamp belongs to the machine with the faster clock rather than to the later write.

Logical clocks give ordering without physical time. A Lamport clock is one counter per node. Send the counter with every message, and set the local counter to one more than the maximum of the local and received values on receipt. The result orders causally related events correctly and gives no way to tell whether two events were concurrent.

A vector clock keeps one counter per node in the system. Comparing two vectors gives three answers: 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 conflicts that need resolution. The cost is a vector that grows with the node count, which is why Dynamo-style systems bound it by pruning old entries.

  node A   [1,0,0] ---- write X=1 ----> [2,0,0]
                \                            \
                 \ replicate                  \ replicate
                  v                            v
  node B   [0,1,0] ---- write X=2 ----> [0,2,0]

  A's version: [2,0,0]      B's version: [0,2,0]
  Neither dominates the other: the writes are CONCURRENT.
  A wall-clock timestamp would silently pick one and drop the other.

  Resolution options: keep both as siblings and let the application
  merge, use a CRDT that merges deterministically, or fail the write.

Hybrid logical clocks, or HLC, combine both. An HLC timestamp holds a physical component and a logical counter. It stays within the clock error of physical time, so it is meaningful to a human, and it respects causality exactly, so it is meaningful to an algorithm. CockroachDB and several others use HLC as their transaction timestamp.

Google TrueTime takes the opposite approach and makes the uncertainty explicit. It returns an interval rather than an instant, with atomic clocks and GPS receivers keeping the interval small. Spanner commits a transaction by waiting out the uncertainty, which is a few milliseconds, so that the commit timestamp is definitely in the past everywhere. The wait buys external consistency, which is linearizability across the whole database including across regions.

Consensus: Paxos and Raft

Consensus means that a set of nodes agrees on one value, with three properties. Agreement: no two nodes decide differently. Validity: the decided value was proposed by someone. Termination: every correct node eventually decides.

Paxos solves it. A proposer picks a proposal number and sends a prepare message to acceptors. An acceptor promises not to accept any lower-numbered proposal and reports the highest-numbered value it already accepted. Once a majority promises, the proposer sends an accept message with either the reported value or its own. A majority of accepts decides the value.

The protocol is correct and it decides one value. Replicating a log needs Multi-Paxos, which elects a stable leader so that the prepare phase runs once rather than per entry. Steady state then costs one round trip to a majority per entry. Paxos is famously hard to translate into an implementation, because the papers describe the safety argument rather than a system.

Raft was designed for understandability and it has become the default. It separates the problem into leader election, log replication, and safety, and it constrains the design so that each part is simple.

  STATES
      +----------+  times out     +-----------+  majority votes  +--------+
      | FOLLOWER |--------------->| CANDIDATE |----------------->| LEADER |
      +----------+  starts vote   +-----------+                  +--------+
           ^                            |                             |
           |  discovers higher term     |  discovers leader or         |
           +----------------------------+  higher term  <-------------+

  STEADY STATE, one client write
    client --> LEADER
                 1. append entry to local log
                 2. AppendEntries to all followers        (parallel)
                 3. wait for a majority to persist        (1 RTT)
                 4. mark committed, apply to state machine
                 5. reply to client
                 6. followers learn commitIndex on the next heartbeat

  Cost: 1 round trip to the majority, plus one fsync on each of
  those nodes. With 5 nodes, the 3rd fastest response sets the latency.

Raft's safety rests on a few rules. A candidate wins only with a majority. A voter grants a vote only if the candidate's log is at least as up to date as its own, which guarantees that a new leader holds every committed entry. A leader never overwrites its own entries, and it never marks an entry from a previous term as committed by counting replicas. That last rule is subtle and it is where naive implementations break.

Membership change is the other subtle part. Moving from three nodes to five in one step can produce two disjoint majorities during the transition, which elects two leaders. Raft solves it with joint consensus, where a transitional configuration requires majorities in both the old and the new set, or with single-server changes, which add or remove one node at a time.

Two refinements reduce these numbers. Flexible Paxos observes that the two phases need only intersecting quorums rather than two majorities, so a system that elects leaders rarely can use a small replication quorum and a large election quorum. Witness replicas hold only metadata and vote, which gives fault tolerance without a full data copy in a third region and cuts cost sharply.

Quorums, leaderless replication, and CRDTs

Quorum replication stores each item on N replicas, writes to W of them, and reads from R. When W plus R exceeds N, every read overlaps every write, so a read sees the latest write. Common values are N equal to 3 with W and R both equal to 2.

Dynamo-style systems make W and R per request. A write with W equal to 1 returns fast and risks loss. A read with R equal to 1 returns fast and risks staleness. The application chooses per call, which is the flexibility that Table 8.2 described.

Leaderless replication needs anti-entropy, because quorum overlap does not repair the replicas that missed a write. Read repair fixes divergence that a read discovers. Hinted handoff stores a write for an unavailable replica on a neighbor and delivers it later. Merkle trees compare large key ranges between replicas with a logarithmic number of comparisons, which makes full reconciliation affordable.

Conflict-free replicated data types, or CRDTs, remove conflict resolution entirely for the data types that admit it. A CRDT has a merge function that is commutative, associative, and idempotent, so replicas that receive the same set of updates in any order converge to the same state. A grow-only counter, a set with add and remove, a last-writer-wins register, and a sequence type for text all exist.

CRDTs power collaborative editors, offline-first mobile applications, and multi-region caches. The limit is expressiveness. Any invariant that spans objects, such as "the total never goes below zero", cannot be a CRDT, because two replicas can each decrement legally and the merged result violates the rule. Where an invariant like that exists, consensus is unavoidable.

Atomic commit across shards

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

Two-phase commit does it. A coordinator sends prepare to every participant. Each participant makes the change durable, holds its locks, and votes. If every vote is yes, the coordinator sends commit. Otherwise it sends abort.

The protocol is correct and it blocks. If the coordinator fails after some participant voted yes, that participant holds its locks and cannot decide alone, because it does not know how the others voted. The locks stay until the coordinator returns. In a database that is a stalled shard, and in a service mesh it is a cascading timeout.

Three-phase commit adds a round to remove the blocking under a synchronous model, and it fails under network partition. The real fix is to make the coordinator fault tolerant by replicating its decision with 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. The coordinator cannot disappear, because it is a replicated group.

Testing a distributed system

Ordinary tests do not find distributed bugs, because the bugs need a specific interleaving of a specific failure. Three techniques do find them.

Fault injection under a consistency checker is the Jepsen method. Generate a concurrent workload, inject partitions, clock skew, and process pauses, record every operation, and then test whether the recorded history is linearizable. Jepsen has found serious correctness bugs in most major distributed databases, and the reports are worth reading as an education in failure modes.

Deterministic simulation runs the whole system in one process with a simulated clock, a simulated network, and a seeded random number generator. A failing seed reproduces the bug exactly, every time, which turns a heisenbug into an ordinary bug. FoundationDB built its reputation on this method, and TigerBeetle, Antithesis, and several others now apply it.

Formal specification checks the protocol rather than the code. TLA+ describes a system as a state machine and a set of invariants, and the model checker explores the reachable states. Amazon Web Services reported using it on S3, DynamoDB, and EBS and finding bugs that testing did not find, including one that needed 35 steps to reach. The specification is not the implementation, so it proves the design and not the code, and the design is where these bugs live.

Sources and further reading

Martin Kleppmann, "Designing Data-Intensive Applications", for the whole territory. Diego Ongaro and John Ousterhout, "In Search of an Understandable Consensus Algorithm", the Raft paper, plus Ongaro's thesis for membership changes. Leslie Lamport, "Paxos Made Simple", and Heidi Howard's work on Flexible Paxos. Daniel Abadi, "Consistency Tradeoffs in Modern Distributed Database System Design", for PACELC. Corbett and others, "Spanner", OSDI 2012. DeCandia and others, "Dynamo", SOSP 2007. Marc Shapiro and others on CRDTs. Chris Newcombe and others, "How Amazon Web Services Uses Formal Methods", CACM 2015. The Jepsen reports at jepsen.io.