09

Consistency and consensus

Replicated systems that promise only eventual consistency leave applications to reason about stale reads. This chapter climbs the ladder of stronger guarantees: linearizability makes many copies look like one, causality and total ordering tell everyone what happened first, and atomic commit makes a transaction land everywhere or nowhere. Each of them turns out to be the same problem in disguise, consensus, which can be solved despite crashes as long as a majority of nodes is alive and connected, and which you will usually borrow from a coordination service rather than build.

Linearizability: one copy, atomic operations

linearizabilityeventual-consistencycompare-and-set
Each bar is a client operation, from the moment the request was sent to the moment the response arrived. The history is linearizable if every operation can be given a single instant inside its bar such that, read left to right, the instants form a legal sequence on one register. The checker searches all orders live.
C writes 1 while A and B poll. A sees the new value, then B, starting strictly later, sees the old one.

History of one register x (initially 0)

✗ not linearizable
020406080100real time →ABCwrite(x, 1)read(x) → 0read(x) → 1read(x) → 1read(x) → 0read(x) → 1

Drag any operation along its row to change when it ran; click a read or CAS to change what it returned. Green ticks are linearization points: the moment each operation appears to take effect atomically. Try: in the first history, drag B's first read left until it overlaps A's second read.

verdict
violation
no order fits both real time and the results
suspect results
2
outlined red: changing any one alone fixes it
results that would fix it
  • A @4: no change here helps
  • A @24: 0
  • A @78: no change here helps
  • B @44: 1
  • B @64: no change here helps

CAP, done right: the choice only bites during a partition

cap-theoremlinearizability-cost
Two datacenters share a register. In linearizable mode every operation goes through the leader in DC1; in available mode each DC serves locally and replicates asynchronously. Random client traffic runs in both.

Two datacenters, one register

t = 0.0 s
datacenter 1x=0leaderdatacenter 2x=0followerWAN · 80 ms
■ ok■ stale read■ error / timeout● replication● client request to leader

Try: partition the WAN in each mode. Linearizable: DC2's clients start timing out, DC1 carries on. Available: everyone keeps working, but DC2 reads stale values until the network heals. Then slide the delay: in linearizable mode DC2 pays 2× it on every operation even with no fault at all.

DC1 availability
100.0%
DC2 availability
100.0%
0 errors
DC1 latency
0 ms
local 2 ms
DC2 latency
0 ms
≈ 2 × WAN delay
stale reads
0
missed an acknowledged write
DC2 behind by
0 ms
replication lag
recent client ops

    Ordering and causality: Lamport timestamps vs version vectors

    causal-consistencylamport-timestampsvector-clocklamport-uniqueness-gap
    Build a history of local writes and messages between three nodes. Every event gets a Lamport timestamp and a version vector, computed by the real update rules. Select an event to see its causal past and future.
    node

    Three nodes exchanging messages

    Acounter 3 · [3,0,0]Bcounter 4 · [2,2,0]Ccounter 4 · [0,0,4]m1→B ⟨4⟩m2→A ⟨4⟩110022001001200230033210400433004220
    number above = Lamport counterdigits below = version vector [A B C]● causal past of the selected event● causal future○ concurrent

    Click events to compare two of them. Try: in the example (preselected), compare C's second write (counter 2) with A's second write (counter 3). Lamport puts C's write first, yet neither could have known about the other; only the version vectors reveal that they are concurrent. Then deliver m1 and m2 and watch the counters jump to max + 1.

    Lamport total order says
    (2,C) < (3,A): C3 first
    version vectors say
    concurrent: neither knew about the other
    total order by (counter, node id)

    Consistent with causality: every event sorts after everything in its causal past. But it is only known after the fact: a node cannot tell whether some other node has issued a smaller timestamp it hasn't heard about yet.

    Lamport timestamps can decide which of two concurrent username claims wins, but only in hindsight. When a request arrives, the node cannot know whether another node is right now processing a claim with a smaller timestamp unless it asks every other node, and if one of them is unreachable, it is stuck. Deciding now needs the order to be final, which is what total order broadcast provides.

    Total order broadcast: same messages, same order, everywhere

    total-order-broadcastlamport-uniqueness-gap
    Pairs of users race to claim the same username at two different replicas. Without an agreed order, each replica believes whichever claim reached it first. With a sequencer, claims become numbered log entries and each replica holds back anything that arrives early.

    Racing username claims, replicated three ways

    t = 0 ms
    N1u1's nodeN2u2's nodeN3u3's node
    ● forwarded claim
    N1 delivery order
    • —
    N2 delivery order
    • —
    N3 delivery order
    • —

    Each claim race sends the same fresh username to two different nodes at nearly the same moment. Try: run plain broadcast and watch replicas crown different owners; switch to total order broadcast and see out-of-order arrivals held back until the gap fills, so every replica applies the same log.

    replicas disagree on
    0 names
    of 0 claimed
    held back
    0
    arrived before a predecessor
    who owns each name?
    nameN1N2N3
    The sequencer here is a single node. If it dies, someone must agree on the next sequencer, and on where its log ended: total order broadcast is equivalent to repeated consensus. Linearizable compare-and-set falls out of it: append your claim, read the log, and the first claim for the name wins on every replica.
    A coordinator runs one transaction across three databases. Step through the protocol, pick a participant's vote, and choose when the coordinator crashes.
    P2 votes
    fault

    Two-phase commit across three databases

    step 0
    applicationcoordinatortransaction managercoordinator log (on disk)P1 · dbidleP2 · dbidleP3 · dbidle
    The application is about to run transaction T, which writes to three different databases.

    Try: with “coordinator dies after votes”, step until the crash and press Wait a few times: the locks never go away. Then make a participant give up, and recover the coordinator with “dies after logging commit” to see why it isn't allowed to.

    outcome
    pending
    in doubt
    0
    prepared, no decision
    DBs holding locks
    0
    steps blocked
    0
    while coordinator is down
    what happened

      Fault-tolerant consensus: a Raft cluster you can break

      consensusconsensus-algorithmsepoch-numberflp-impossibility
      Five nodes run leader election with randomized timeouts and replicate a log with AppendEntries. An entry commits only once a majority of nodes have stored it in the leader's term. Crash nodes and cut the network to find the edges of what a majority can do.

      Five Raft nodes

      t = 0.00 s
      N1term 0N2term 0N3term 0N4term 0N5term 0
      logs · color = term · solid = committed
      N1
      N2
      N3
      N4
      N5
      1
      2
      3
      4
      5
      6
      7
      8
      9
      10
      11
      12
      13
      14
      events
      • press Play: every node starts as a follower in term 0
      ● RequestVote / vote● AppendEntries (big = carries entries, small = heartbeat)grey arc = election timerclick a node to crash / restart it

      Try: let a leader emerge, write a few entries, then partition the leader with one follower. The old leader keeps accepting writes it can never commit (outlined); the majority side elects a new leader in a higher term. Heal, and the old leader steps down and its uncommitted entries are overwritten. Then crash three nodes: nothing commits without a majority.

      leader
      none
      election in progress
      highest term
      0
      0 elections started
      committed entries
      0
      safe on a majority
      nodes up
      5/5
      quorum needs 3
      Compare with 2PC: there the coordinator's death blocks everyone and every participant must vote yes. Consensus needs only a majority, and if the leader dies, the others time out, elect a successor in a higher term, and continue. The price is that a minority cannot make progress at all, and that elections stall when timeouts misfire on a flaky network.

      Membership and coordination services

      zookeeper-coordinationephemeral-znode
      Three clients compete for a lock on a ZooKeeper-like service: each creates an ephemeral sequential znode, the lowest number holds the lock, and sessions stay alive only while heartbeats arrive. The holder writes to a storage service using its znode's zxid as a fencing token.
      storage checks fencing tokens

      A lock on a coordination service

      t = 0.0 s
      ZooKeeper ensemble
      3 nodes, writes via a consensus log · zxid 33
      /lock
      • n-0001 A zxid 31 ← lock
      • n-0002 B zxid 32
      • n-0003 C zxid 33
      sessions
      A
      B
      C
      client Arunning
      believes it holds the lock · token 31
      client Brunning
      waiting on predecessor's znode
      client Crunning
      waiting on predecessor's znode
      storage service
      highest token seen: —
        • 0.0s A is notified: it now holds the lock (token 31)
        • 0.0s C creates ephemeral /lock/n-0003 (zxid 33)
        • 0.0s B creates ephemeral /lock/n-0002 (zxid 32)
        • 0.0s A creates ephemeral /lock/n-0001 (zxid 31)

        Try: press Play, then GC-pause the lock holder for longer than the 3-second session timeout. Its session expires, the ephemeral znode vanishes and the next client is notified. Resume it: it still thinks it holds the lock. With fencing on, its stale token is rejected; turn fencing off and repeat to see the corrupt write.

        lock holder (per ZooKeeper)
        A
        token 31
        clients who think they hold it
        1
        stale writes rejected
        0
        corrupting writes
        0
        • takeawayLinearizability is a recency guarantee: once a new value is visible to anyone, it is visible to everyone. It is invaluable for locks, leader election and uniqueness, but every operation costs a round trip, and a partitioned minority must refuse to answer.
        • takeawayCausality is a partial order that can be tracked cheaply (Lamport timestamps, version vectors). Deciding things the moment they happen needs a total order everyone agrees on, which means total order broadcast, which is consensus.
        • takeaway2PC is consensus that blocks when the coordinator dies. Raft, Paxos and Zab survive any minority of failures by using epochs and overlapping majority quorums. Most applications should borrow them through ZooKeeper or etcd rather than reinvent them.

        Key concepts · 20

        book chapter 9
        1. Eventual consistency

          If writes stop, all replicas eventually return the same value; until then, reads may return anything older.

          It is a very weak guarantee: it says nothing about when replicas converge, and it makes reads after writes unpredictable. Stronger models cost performance, which is why many systems settle for it.

          Dynamo-style storesasync followers
          ↑ see it in the visualization
        2. Linearizability

          Make a replicated object behave as if there were one copy and every operation took effect atomically at a single point between its start and end.

          It is a recency guarantee: once any read returns a new value, all later reads (in real time) must too. Checking it means finding a total order of operations that respects real time and register semantics.

          etcdZooKeeper writesSpanner
          ↑ see it in the visualization
        3. Compare-and-set

          An atomic operation that sets a value only if it currently equals an expected value, and reports whether it succeeded.

          With a linearizable register plus CAS you can implement locks, leader election and uniqueness constraints; whether a CAS may succeed depends on the single global order of all writes.

          ↑ see it in the visualization
        4. CAP theorem

          During a network partition, a system must choose between linearizability and availability.

          Partitions are not optional, so CAP really says: when one happens, either some clients get errors (stay consistent) or everyone gets answers that may be stale (stay available). It is a narrow theorem, more slogan than design tool.

          ↑ see it in the visualization
        5. Cost of linearizability

          Linearizable reads and writes have response times at least proportional to network delay.

          Most systems that drop linearizability (even multicore CPU caches) do it for speed, not for partition tolerance. On a WAN with variable delays, every linearizable operation pays the round trip.

          ↑ see it in the visualization
        6. Causal consistency

          Operations that are causally related are seen by everyone in the same order; concurrent ones may be seen in any order.

          Causality gives a partial order, linearizability a total one. Causal consistency is the strongest model that does not slow down with network delay and stays available during partitions.

          ↑ see it in the visualization
        7. Lamport timestamps

          A (counter, node id) pair; every node sends its counter on every message and, on receipt, sets its counter to max(local, received) + 1.

          They give a compact total order that is consistent with causality: if A happened before B, A's timestamp is smaller. The converse fails: a smaller timestamp does not mean the event came first causally.

          ↑ see it in the visualization
        8. Version vectors (vector clocks)

          One counter per node, merged element-wise on each message, so each event carries a summary of everything it has seen.

          Unlike Lamport timestamps, vectors can tell whether two events are concurrent or causally ordered, at the cost of size growing with the number of nodes.

          RiakDynamo
          ↑ see it in the visualization
        9. Why timestamps aren't enough

          A total order of timestamps can pick a winner among conflicting claims only after all claims have been collected.

          To enforce 'one owner per username' at the moment of the request, a node must know no other node is concurrently claiming it with a lower timestamp, which requires hearing from every node. You also need to know when the order is final: total order broadcast.

          ↑ see it in the visualization
        10. Total order broadcast

          A protocol that delivers every message reliably to every node, and in the same order on all of them.

          It is state machine replication: apply the same deterministic operations in the same order and replicas stay identical. It is equivalent to repeated consensus, and lets you build linearizable compare-and-set on top of an append-only log.

          ZooKeeper Zabetcd RaftKafka (per partition)
          ↑ see it in the visualization
        11. Atomic commit

          Ensuring that a transaction spanning several nodes either commits on all of them or aborts on all of them.

          Committing independently on each node risks some committing while others abort, violating atomicity. Once committed, a transaction cannot be retracted, so the decision must be made together.

          ↑ see it in the visualization
        12. Two-phase commit (2PC)

          A coordinator asks every participant to prepare (vote), then writes a commit or abort decision to its log and broadcasts it.

          Two points of no return make it work: a participant that votes yes can no longer abort on its own, and the coordinator's logged decision is final. Its weakness is that it blocks if the coordinator dies.

          ↑ see it in the visualization
        13. In-doubt transaction

          A participant that has voted yes but not yet heard the decision: it can neither commit nor abort.

          It must hold its locks until the coordinator recovers, blocking other transactions. Orphaned in-doubt transactions sometimes need a human (heuristic decision) to resolve, which risks breaking atomicity.

          ↑ see it in the visualization
        14. XA transactions

          A standard C API for two-phase commit across heterogeneous systems such as databases and message brokers.

          The coordinator is often a library inside the application, so if the application server dies, its log goes with it. XA must also be lowest common denominator: it cannot detect deadlocks across systems or support SSI.

          JTAMySQL XAPostgreSQL PREPARE TRANSACTION
        15. Consensus

          Getting several nodes to agree on a value, with uniform agreement, integrity, validity and termination.

          Leader election, atomic commit, uniqueness, linearizable CAS and total order broadcast all reduce to it. Unlike 2PC, fault-tolerant consensus keeps making progress as long as a majority is up.

          ↑ see it in the visualization
        16. Fault-tolerant consensus algorithms

          Algorithms such as Raft, Multi-Paxos, Zab and Viewstamped Replication that decide a sequence of values (a log).

          They elect a leader per epoch and require a majority quorum both to elect it and to accept each proposal; any two majorities overlap, so a decision can never be lost or contradicted.

          RaftPaxosZabVSR
          ↑ see it in the visualization
        17. Epoch (term) numbers

          A monotonically increasing number attached to each leadership; a higher epoch always wins.

          A leader cannot trust that it is still leader. Before deciding anything it needs a quorum of acknowledgments in its epoch, so a deposed leader from an older epoch can never get a value committed.

          ↑ see it in the visualization
        18. FLP impossibility

          In a fully asynchronous model with no clocks, no deterministic algorithm can always reach consensus if a node may crash.

          Real systems escape it with timeouts (to suspect crashed nodes) or randomness. That is why election timeouts are randomized and why consensus can stall when the network is very unreliable.

          ↑ see it in the visualization
        19. Coordination service

          A small, fault-tolerant, linearizable key-value store that applications use for locks, leader election, membership and configuration.

          It outsources consensus: few apps should implement Raft themselves. It offers linearizable atomic operations, total ordering (zxid as fencing tokens), failure detection via sessions, and change notifications.

          ZooKeeperetcdConsul
          ↑ see it in the visualization
        20. Ephemeral nodes and sessions

          A znode tied to the client session that created it; when heartbeats stop and the session times out, it is deleted automatically.

          It turns failure detection into data: a lock or membership entry disappears when its owner dies. But a paused client may not know its session expired, so the resources it guards must check fencing tokens.

          ZooKeeperetcd leases
          ↑ see it in the visualization