Consistency and Consensus

An interactive walk through Chapter 10 of Designing Data-Intensive Applications (2nd ed.). It covers what “strong consistency” precisely means, why it’s expensive, how to generate IDs and timestamps that respect causality, and how consensus algorithms keep a replicated system correct through all the faults of chapter 9. The centrepiece is a five-node consensus cluster you can break.

1The big picture

Replication is one of our best tools for fault tolerance, but multiple copies invite inconsistency: stale reads from lagging replicas, and conflicts when several replicas accept writes. There are two philosophies:

This chapter explores the strong side in three steps: a precise definition (linearizability), the surprisingly related problem of generating IDs and timestamps, and how to get linearizability with fault tolerance: consensus. The book stays informal and skips proofs; real consensus work needs the theory behind it, because these systems tend to work fine until an unlucky combination of faults and message orderings. The chapter’s epigraph says it well: “Never go to sea with two chronometers; take one or three.”

Chapter map

2Linearizability

The idea: as if there were one copy

The simplest replicated database to use behaves like a single-node database: no replication lag, no conflicts. That is linearizability, also called atomic, strong, immediate or external consistency. The system appears to have one copy of the data, and every operation on it is atomic. As soon as one client’s write completes, every client’s read must see it, never a value from a stale cache or replica. In short, it is a recency guarantee.

What makes a system linearizable?

Theory calls the object a register (in practice a key, row or document). Each request is a bar from when the client sent it to when the response arrived. The client only knows the database processed it somewhere inside that bar. Operations: read(x) ⇒ v, write(x, v) ⇒ ok, and cas(x, v_old, v_new) ⇒ ok/error.

• A read that finishes before a write starts must return the old value; one that starts after the write finishes must return the new one.
• A read that overlaps the write may return either (Figure 10-2).
• But there must be a single moment at which the value flips: once any read has returned the new value, every later read (by anyone) must too, even if the write hasn’t finished (Figure 10-3).
• Formally: you can mark a point inside each bar where it “took effect”, joining the points gives a valid sequence for one register, and the joining line never goes backwards in time (Figure 10-4). Requests may take effect in a different order than they were sent, as long as they were concurrent.

Checking whether recorded behaviour is linearizable is possible but computationally expensive; the lab does it by brute-force search. Weaker models also exist, just as weaker isolation levels do: read-after-write consistency, monotonic reads and consistent prefix reads (chapter 6) are all implied by linearizability, the strongest model in common use.

Linearizability vs serializability

Both together are strict serializability (strong one-copy serializability, strong-1SR). Single-node databases are typically linearizable. CockroachDB offers serializability with some recency guarantees but not strict serializability, which would need expensive coordination; Spanner and FoundationDB are strictly serializable. Isolation level and consistency model can mostly be chosen independently. (Sequential consistency is yet another model, not covered here.)

Relying on linearizability

A slightly stale sports score harms nobody. But some things only work with linearizability:

Cross-channel timing dependencies

Bryce only noticed the stale result because Aaliyah told him: a second channel, outside the database. Computer systems have such side channels too:

Linearizability is the simplest fix. If you control the second channel (unlike Aaliyah’s voice), read-your-writes-style techniques also work, at the cost of extra complexity.

Implementing linearizable systems

The simplest way to act like one copy is to have one copy, but then there’s no fault tolerance. How do the replication methods from chapter 6 fare?

The cost of linearizability

Multi-leader replication suits multi-region deployments. What happens when the link between regions breaks, while each region keeps working internally?

This isn’t specific to single- vs multi-leader, or to multi-region setups. Any linearizable database on an unreliable network faces it:

This is the CAP theorem, named by Eric Brewer in 2000, though designers knew the trade-off since the 1970s. It started as a rule of thumb, and it deserves credit for pushing engineers beyond shared-storage clusters toward shared-nothing designs, which helped trigger the NoSQL movement. Formally it is very narrow: one consistency model (linearizability) and one fault (network partitions, under 8% of incidents according to Google’s data). It says nothing about delays, dead nodes or other trade-offs. PACELC adds: if Partitioned, choose A or C; Else, choose Latency or Consistency, but it inherits CAP’s odd definitions. More precise impossibility results have superseded CAP, so it’s mostly of historical interest.

The unhelpful CAP theorem

Linearizability and network delays

Surprisingly few systems are linearizable, not even RAM on a multi-core CPU. Each core has its own cache and store buffer, writes reach main memory asynchronously, and another core may read an old value unless a memory barrier (fence) is used. Nobody blames CAP for that: cores don’t keep running when disconnected. The reason is performance. The same holds for most databases that drop linearizability: it’s about speed, because linearizable systems are slower all the time, not only during faults.

Chapter 13 discusses ways to avoid linearizability without sacrificing correctness.

3ID generators and logical clocks

Records need unique IDs. Single-node databases usually use an autoincrementing integer: compact (64 bits, or 32 if you dare), and its order tells you creation order, so a chat app can sort messages by ID and replies follow questions. That generator is itself linearizable: each request is an atomic fetch-and-add, so if Aaliyah’s post finishes before Bryce’s starts, Bryce gets the bigger ID. Concurrent posts may come out either way. Easy in memory with a CPU atomic instruction; it has to be persisted so that a restart doesn’t issue duplicates. The real problems:

• it’s a single point of failure
• it’s slow from other regions (a round trip across the planet for an ID)
• it can become a bottleneck at high write throughput

Distributed alternatives

Atomic clocks and GPS shrink the problem, but it would be nicer to get correctly ordered unique IDs without special hardware.

Logical clocks

Time-of-day and monotonic clocks are physical clocks: they measure elapsed time. A logical clock is an algorithm that counts events. Its timestamps don’t tell you the time, but they can be compared. Requirements:

• timestamps are compact (a few bytes) and unique
• any two can be compared: they are totally ordered
• the order is consistent with causality: if A happened before B, A’s timestamp is smaller

A single-node counter meets these; the distributed schemes above fail the causality requirement.

Lamport timestamps

Leslie Lamport’s 1978 clock (one of the most-cited papers in distributed systems) is a pair (counter, node ID). Generating a timestamp increments your counter; seeing someone else’s timestamp with a bigger counter bumps yours up to match. Compare by counter first, then node ID. It gives a total order consistent with causality, but it is not linearizable: it doesn’t make anything up to date, it just orders IDs.

Hybrid logical clocks

Lamport counters have no relation to real time (you can’t find “all messages posted on Tuesday”), and nodes that never talk drift far apart in counter values. A hybrid logical clock counts physical time (seconds or microseconds) but, like Lamport, jumps forward when it sees a larger timestamp, and increments every time it generates one, so it moves forward even if NTP steps the physical clock backwards. It stays close to physical time, needs only rough clock sync and no special hardware, and its order respects happens-before. CockroachDB uses them. Try HLC in the ID lab above with node B’s clock running behind.

Lamport and hybrid clocks vs vector clocks

Both make good MVCC transaction IDs for snapshot isolation, because snapshots then respect causality. But they order concurrent events arbitrarily, so from two timestamps you generally can’t tell whether the events were concurrent. (Equal counters do reveal concurrency; different counters don’t prove causality.) To detect concurrency you need vector clocks: one counter per node, stored with every write. If A is ahead for one node and B for another, they’re concurrent. The price is size: potentially one integer per node in the system. Toggle the vectors in the lab above to compare.

Linearizable ID generators

Lamport and hybrid clocks only guarantee a node’s timestamps exceed those it has seen. Linearizability is stronger: if A completed before B began, B gets the higher ID even if A and B never communicated. That difference can leak data:

Other fixes are fragile: the photos DB could check the account first (easy to forget), or the app could track the user’s latest timestamp (hard across a laptop and a phone). A linearizable ID generator is the simplest.

Implementing a linearizable ID generator

• One node: atomically increment and return a counter, persist it so restarts don’t duplicate IDs, and replicate it with single-leader replication. TiDB/TiKV call this a timestamp oracle (inspired by Google’s Percolator).
• Batching: instead of a disk write and replication per ID, persist and replicate “IDs up to N are reserved”, hand those out from memory, and reserve the next batch before running out. A crash or failover skips some IDs, but there are never duplicates or out-of-order IDs.
• It can’t be sharded (independent shards break the order) or easily spread across regions, so in a geo-distributed database every ID request goes to one region. Luckily the job is so simple that one node handles a lot of throughput.
• Spanner’s alternative: clocks that report an uncertainty interval, plus waiting out that interval before returning (chapter 9). If the interval is correct, a request that starts after another finishes gets a larger timestamp, with no communication at all, even across regions. It needs tightly synchronized clocks and support for computing uncertainty.

Enforcing constraints using logical clocks?

Could a linearizable clock decide who gets a lock or a username, with the lowest timestamp winning? Future requests will always get larger timestamps, so a winner can’t be overtaken later. But a node still has to know its timestamp is the lowest right now:

4Consensus

Several things in this chapter are easy on one node and hard with fault tolerance:

• A single leader is linearizable, but how do you fail over without split brain, and make sure a paused “leader” hasn’t been voted out meanwhile?
• A linearizable ID generator is just an atomic counter, until its node crashes.
• A CAS decides who gets a lock or a unique name, and on one node it’s a single CPU instruction. How do you make it fault-tolerant?

They are all the same fundamental problem: consensus, getting several nodes to agree on a value. It’s one of the most important problems in distributed computing and notoriously easy to get wrong. The best-known algorithms are Viewstamped Replication, Paxos, Raft and Zab: similar, not identical, and all non-Byzantine (messages may be delayed or lost and nodes may crash, but nobody lies). Byzantine-tolerant consensus (usually assuming fewer than a third of nodes are faulty) is used in blockchains and is out of scope.

The impossibility of consensus (FLP)

The FLP result (Fischer, Lynch, Paterson) proves that no algorithm can always reach consensus if a node may crash. So why do consensus algorithms exist? Because FLP assumes a deterministic algorithm in the asynchronous model, with no clocks and no timeouts. Allow timeouts to suspect crashed nodes (even if the suspicion is sometimes wrong), or allow random numbers, and consensus becomes solvable. FLP matters in theory; in practice consensus is usually reached.

The many faces of consensus

These are all equivalent: a solution to one can be converted into a solution to any other, which is why they’re all called “consensus”. (One caveat for fetch-and-add, below.) See how the same goal is reached with each primitive:

• CAS ⇄ consensus: start the register at null; everyone CASes from null to their proposal; decide what the register holds. Conversely, run consensus among CAS calls with the same expected value; the decided one succeeds and the others fail. Object stores’ conditional writes are distributed CAS.
• Shared log ⇄ consensus: everyone appends their proposal; decide the first entry. Conversely: one consensus instance per log slot. Propose your value for an undecided slot, append decided values once all earlier slots are decided, and retry in a later slot if you lose. Single-leader replication without failover is a log but fails liveness, since it stops when the leader dies.
• Fetch-and-add: easy to build from CAS (read, CAS old→old+1, retry). The other way round only works for 2 proposers: its consensus number is 2, while CAS and logs have ∞.

Single-value consensus

One or more nodes propose values; the algorithm decides one, for example who becomes leader, gets the lease, or buys the last seat. For several decisions (one per theater seat), run one instance each. Required properties:

Without fault tolerance the first three are trivial: make one node the dictator. All the difficulty is in termination: decide even if a node crashes and never returns (assume an earthquake buried its datacenter under 30 feet of mud). If all nodes crash nothing can be decided; it’s provable that termination needs a majority of nodes working, and that majority forms a quorum. Most algorithms keep the safety properties even if a majority fails: an outage can stop progress, but never cause inconsistent decisions.

Each “algorithm” below is broken. Which property does it violate?

Atomic commitment as consensus

Two-phase commit (chapter 8) also gets nodes to agree, with one difference: consensus may decide any proposed value, but atomic commit must abort if anyone votes abort.

From consensus: everyone sends their vote to everyone; a node that received “commit” from all proposes commit, while anyone who saw an abort or a timeout proposes abort; all follow the consensus decision. (If all voted commit but something timed out, it doesn’t matter which way it goes, as long as everyone agrees.) Back again: to propose a value, run a transaction on a quorum that CASes a register if unset, voting commit if the CAS succeeded; a committed transaction decides its value, an aborted one retries. Equivalent once more. Unlike 2PC, a consensus-based commit doesn’t hang on a single coordinator.

5Consensus in practice

All the formulations are equivalent, but which do systems actually provide? Mostly shared logs. Raft, Viewstamped Replication and Zab provide a log out of the box; Paxos decides one value, but most deployments use Multi-Paxos, which provides a log.

Using shared logs

• State machine replication: if each entry is a write and every replica applies the same entries in the same order with deterministic logic, all replicas end up identical. This is the principle behind event sourcing, and logs are central to stream processing (chapter 12).
• Serializable transactions: make each entry a deterministic stored-procedure transaction executed in log order (as in actual serial execution).
• Sharded strongly consistent databases often keep one log per shard, which scales but limits cross-shard guarantees (consistent snapshots, foreign keys); cross-shard serializability needs extra coordination.
• Other forms of consensus from a log: CAS = first entry wins; many single-value decisions (one per seat) = first entry mentioning that seat; fetch-and-add = put increments in entries and sum them. A simple per-entry counter yields fencing tokens: ZooKeeper’s zxid.

From single-leader replication to consensus

A single “dictator” makes consensus easy, and a single leader makes a log easy. The question is safe failover. Traditional databases left it to a human, which means downtime and doesn’t satisfy termination. Automatic election is tricky: you seem to need consensus to elect a leader, and a leader to do consensus.

The way out: algorithms don’t guarantee one leader at any moment, only one leader per epoch. The epoch is called a ballot number in Paxos, a view number in Viewstamped Replication, and a term number in Raft.

1. A node that hasn’t heard from the leader for a timeout starts an election with a higher epoch number. If two leaders from different epochs conflict (the old one wasn’t dead after all), the higher epoch wins.
2. Before appending an entry, the leader collects votes from a quorum (typically a majority). A node votes yes only if it knows of no leader with a higher epoch.
3. So there are two rounds of voting, one to elect a leader and one per proposal, and their quorums must overlap. If a proposal vote passes without anyone reporting a higher epoch, no newer leader can exist, so appending is safe.

Subtleties, and a simulator

Raft, Multi-Paxos, VR and Zab share this structure: a quorum elects a leader, and every entry needs another quorum vote. Each entry is replicated synchronously to a quorum before the client is told it succeeded, so it survives the leader failing. The devil is in the details, such as how a new leader honours entries the old leader already appended. Raft only lets a node become leader if its log is at least as up to date as a majority’s. Paxos lets any node win but makes it catch up its log before appending anything new. The simulator follows Raft:

Things to try: crash the leader and watch a new term begin; partition the leader into the minority and send it a write (it’s appended but never commits, and is later overwritten); run the stale-node scenario with unclean election off and then on; cut one link with pre-vote off and then on.

Consistency vs availability in leader election

Strict consensus requires a new leader to be up to date with every confirmed entry before it handles writes or linearizable reads. Otherwise a stale leader could write new values into slots the old leader already filled, breaking the log’s append-only property. Some systems weaken this to recover faster, or to recover at all: Kafka’s unclean leader election lets any replica lead, and databases with asynchronous replication can’t guarantee any follower is current. You gain availability and performance but you are on thin ice: consensus theory no longer applies, and chapter 9’s faults can lose or corrupt data (see the simulator). There’s also subtlety around entries the old leader proposed whose vote never finished.

• Linearizable reads must also go through a quorum, to confirm the node answering is still leader and up to date. etcd works this way.
• Reconfiguration: standard algorithms assume a fixed voting membership; extensions add and remove nodes, which is useful for adding regions or migrating (add the new nodes, then remove the old).

Pros and cons of consensus

6Coordination services

ZooKeeper, etcd and Consul look like key-value stores but are built to coordinate another distributed system: Kubernetes relies on etcd, while Spark and Flink in high-availability mode rely on ZooKeeper. They hold small amounts of data that fit in memory (still persisted to disk), replicated with consensus. They’re modelled on Google’s Chubby and combine consensus with features that turn out to be very handy:

Managing configuration

Timeouts, thread-pool sizes and similar settings can live in a coordination service: processes load them at startup and subscribe to changes, then apply them or restart. This doesn’t need consensus, and polling a file or URL works too, but if you’re running the service anyway its notifications are convenient.

Allocating work to nodes

Typical jobs: choose a leader or primary among instances (for databases, job schedulers and other stateful services) and fail over when it dies; and assign shards to nodes, moving them when nodes join, leave or fail. These are built from atomic operations, ephemeral nodes and notifications. It isn’t easy even with recipe libraries like Apache Curator, but far better than writing consensus yourself. A coordination service runs on a fixed 3 or 5 nodes however big the system it coordinates, so you outsource consensus instead of running it across thousands of nodes. Its data changes slowly (“the node at 10.1.1.23 leads shard 7”, minutes to hours); fast-changing state belongs in a database, or in tools like Apache BookKeeper.

Service discovery

In clouds where VMs come and go, services register their endpoints on startup so others can find them, and a coordination service’s failure detection and notifications make tracking them easy. If you already use one for leases or election, it also knows who should receive requests. But consensus is usually overkill here: discovery doesn’t need linearizability, and it must be highly available and fast or everything stalls. So cache it: retry with fresh data when a connection fails, refresh on a TTL. DNS-based discovery uses layers of caching. ZooKeeper offers observers for this: replicas that receive the log but don’t vote, with possibly stale but always-available reads and more read throughput.

7Summary

Linearizability makes replicated data look like one copy with atomic operations. It’s needed when data must be up to date when read, or to settle races (two people creating the same file name). It’s easy to understand, like a variable in a single-threaded program, but slow, especially with large network delays, and many replication schemes that look strongly consistent don’t provide it.

ID generators: a single-node counter is linearizable but not fault-tolerant; most distributed schemes don’t order IDs consistently with real events. Lamport and hybrid logical clocks order consistently with causality but aren’t linearizable; vector clocks can detect concurrency.

Consensus makes fault-tolerant linearizable replication possible: nodes agree on a single sequence of operations despite delays and failures, so a group behaves like one node processing operations in order. Raft and Paxos are essentially single-leader replication with built-in election and failover, designed so that no committed write is lost and split brain is impossible. That requires a quorum (usually a majority) to confirm every write and every linearizable read, which is expensive across regions but unavoidable for strong consistency with fault tolerance.

Problems that are really consensus

All are easy if a single node decides, which is exactly why single-leader databases can offer them. When that leader fails or becomes unreachable, progress stops until a human fails over, unless you use consensus. If you need any of these to be fault-tolerant, use a coordination service like ZooKeeper or etcd: it won’t guarantee you get it right, but it will probably help. Consensus rests on decades of theory that lets systems tolerate chapter 9’s faults without corrupting data, but it isn’t always the right tool. When strong consistency isn’t needed, leaderless or multi-leader replication with weaker consistency gives better availability and performance, and logical clocks help there.

Self-check quiz

Glossary