8. Distributed Systems Theory

A distributed system is one where machines that fail independently must agree on things, over a network that can delay, reorder, or drop what they say to each other. This chapter covers the results that bound what is achievable — CAP, PACELC, FLP — the consistency models that describe what a system actually promises, the mechanisms for ordering events without a shared clock, and the consensus and transaction protocols that make agreement possible anyway.

8.1 Fundamental Constraints

Before studying any protocol, it is worth knowing which problems are hard by nature rather than by poor engineering. The results in this subchapter are not obstacles to be optimized away — they define the shape of every solution that follows.

Fallacies of Distributed Computing

In 1994, engineers at Sun Microsystems catalogued the assumptions that developers new to distributed systems make without noticing. Each one is false, each one is comfortable, and each one produces a specific class of production incident. The list has aged remarkably well because the fallacies are not about technology — they are about the difference between a function call and a message sent to another machine.

The root cause is that distributed systems are usually presented through abstractions that look local. An RPC call looks like a method call. A remote object looks like an object. The abstraction is useful right up to the moment it fails, and then every difference it was hiding becomes your problem at once.

#	Fallacy	Reality
1	The network is reliable	Packets drop; connections reset; NICs fail
2	Latency is zero	Cross-region round trips are ~100 ms, immovable
3	Bandwidth is infinite	Payload size becomes a scaling constraint
4	The network is secure	Anything on the wire is readable and forgeable
5	Topology doesn't change	Instances move constantly; addresses are temporary
6	There is one administrator	Many teams, many change windows, no single view
7	Transport cost is zero	Serialization, TLS, and egress all cost real money
8	The network is homogeneous	Different MTUs, protocols, proxies, and middleboxes

# The abstraction that hides all eight fallacies at once.
user = user_service.get_user(user_id)      # looks like a local method call
#
# What it actually does: DNS resolution (may be stale), TCP handshake (may
# time out), TLS negotiation (certs may have expired), serialization (schema
# may have changed), transmission (may be dropped), remote execution (may be
# overloaded), response transmission (may be lost after the work was done).
#
# Any of these can fail, and one of them -- a lost response -- is
# indistinguishable from a lost request. That single ambiguity is the source
# of most of this chapter.

# Writing it with the fallacies in view:
try:
    user = user_service.get_user(
        user_id,
        timeout=0.5,                      # #2: latency is never zero; bound it
        retries=2,                        # #1: the network is unreliable
        idempotency_key=request_id,       # a retry may duplicate the effect
        fields=["id", "name", "tier"],    # #3: do not fetch what you will not use
    )
except ServiceUnavailable:
    user = cached_user(user_id) or DEGRADED_USER   # #1: plan for absence

The most consequential fallacy in practice is the first, and its sharpest form is this: a timeout tells you nothing about what happened. When a call times out, the request may never have arrived, may have arrived and be executing now, may have completed successfully with a lost response, or may have failed. The caller cannot distinguish these cases, ever, and the entire discipline of idempotency exists because of it.

 WHAT A TIMEOUT ACTUALLY MEANS

  caller                              callee
    |---- request ---------X          (a) never arrived      -> safe to retry
    |---- request ------------->      (b) still running      -> retry duplicates
    |---- request ------------->      (c) done, reply lost   -> retry duplicates
    |<---------X                      (d) failed remotely    -> safe to retry
    |
    ? All four look IDENTICAL to the caller: no response before the deadline.

Fallacy 2 deserves a number attached to it, because latency is the one constraint that no amount of money removes. Light travels roughly 200 km per millisecond in fibre, so a New York to London round trip cannot go below about 56 ms, and in practice runs 70-90 ms. That figure decides whether a design that requires three sequential cross-region round trips is viable before anyone writes code.

Key Takeaways

The eight fallacies are assumptions that feel true locally and are false the moment a network is involved.

Abstractions that make remote calls look local hide the failure modes until they occur, all at once.

A timeout is ambiguous by nature: four different outcomes are indistinguishable to the caller.

Latency is bounded by physics, so cross-region round-trip counts are a design constraint, not an optimization target.

🧪 Practice

For each of the eight fallacies, name one concrete production incident it could cause.

Compute the minimum theoretical round trip between Sydney and London, then explain why the real number is roughly double.

Interview: A payment call times out. What do you do next, and what must be true of the payment API for your answer to be safe? (Hint: you cannot learn what happened from the timeout — so what must the callee provide?)

CAP Theorem

The CAP theorem, proved by Gilbert and Lynch in 2002, states that a distributed data store cannot simultaneously provide all three of:

Consistency (C): every read receives the most recent write or an error. (This is linearizability, not the "C" of ACID — a persistent source of confusion.)
Availability (A): every request to a non-failing node receives a non-error response.
Partition tolerance (P): the system continues to operate despite messages being dropped between nodes.

The common phrasing — "pick two of three" — is misleading, and understanding why is the whole value of the theorem. Partitions are not a choice. Networks partition; cables are cut, switches fail, a routing change isolates a rack. A system that gives up P is a system that stops working when the network misbehaves, which is not an engineering position anyone holds deliberately.

So the real statement is narrower and far more useful: when a partition occurs, you must choose between consistency and availability. There is no third option, because the choice is forced by the situation itself.

 THE FORCED CHOICE DURING A PARTITION

              ##### network partition #####
   +----------------+          |         +----------------+
   |    node A      |    X-----X-----X   |    node B      |
   |  balance: 100  |          |         |  balance: 100  |
   +----------------+          |         +----------------+
          ^                                      ^
     client 1 writes                        client 2 reads
     "withdraw 100"

  CP choice: B refuses the read (or A refuses the write). No node serves a
             request it cannot confirm is current. Correct, but UNAVAILABLE.

  AP choice: both nodes serve their requests. B returns 100 -- which is now
             stale. Available, but INCONSISTENT, and the two sides must be
             reconciled once the partition heals.

The choice is driven entirely by what a wrong answer costs in that domain:

Domain	Choice	Reasoning
Bank ledger, inventory	CP	A double-spend is worse than a brief error
Shopping cart	AP	A lost cart item is worse than a failed page
Configuration store	CP	Divergent config causes correlated failures
Social feed, likes	AP	Stale counts are invisible; downtime is not
Service registry	AP	A stale instance list beats no list (Chapter 6)

# The choice, made explicit at the point where it is forced.
def read_balance(account_id, quorum_reachable):
    if quorum_reachable:
        return replica.read(account_id, consistency="QUORUM")

    # The partition is here. Choose, per operation -- this is not a
    # system-wide setting in any well-designed system.
    if OPERATION_MODE == "CP":
        raise ServiceUnavailable("cannot guarantee a current read")
    else:  # AP
        value = replica.read(account_id, consistency="ONE")
        return Stale(value, as_of=replica.last_sync_time)   # label it stale so
                                                            # the caller can
                                                            # decide what to do

Three refinements separate a working understanding of CAP from a slogan.

The choice is per operation, not per system. The same database can serve reads of a product description with AP semantics and a stock decrement with CP semantics. Systems that force one global setting are limiting you unnecessarily.

CAP is only about partitions. It says nothing about the normal case, which is 99.9% of the time. A system's everyday consistency and latency behaviour is described by PACELC, the next topic.

"CA" systems do not exist in a distributed setting. A single-node database is CA in a trivial sense — no partitions are possible inside one machine — but the moment there are two nodes, P is mandatory. Vendors claiming CA are describing behaviour in the absence of the only condition the theorem is about.

Key Takeaways

CAP's "pick two" framing misleads: partition tolerance is mandatory, so the real choice is C or A during a partition.

The C in CAP is linearizability, not ACID's consistency.

Choose per operation based on the cost of a wrong answer versus the cost of an error.

CAP says nothing about normal operation, and no genuinely distributed system is "CA".

🧪 Practice

Classify as CP or AP and justify: DNS, a bank transfer, a like counter, a distributed lock, a product catalogue.

Explain in two sentences why "we chose CA" is not a coherent design statement for a multi-node system.

Interview: Your system is CP and a partition takes down a data centre for 20 minutes. What do users experience, and what would AP have cost you instead? (Hint: describe both the outage and the reconciliation you avoided.)

PACELC Theorem

CAP describes behaviour during partitions, which are rare. PACELC, formulated by Daniel Abadi in 2010, extends it to cover the case that is true almost all of the time — the network working normally — and in doing so it explains far more of a system's observed behaviour.

The formulation reads: if there is a Partition (P), choose between Availability (A) and Consistency (C); Else (E), choose between Latency (L) and Consistency (C).

The second half is the contribution. Even with a perfectly healthy network, a system that guarantees consistency must coordinate before answering — contact a quorum, wait for a leader, confirm replication — and coordination takes time. A system willing to return a possibly-stale local value answers immediately. That trade exists constantly, not only during incidents, and it dominates the latency users actually experience.

 THE EVERYDAY TRADE (no partition in sight)

 CONSISTENT READ                        LOW-LATENCY READ

  client -> replica in eu-west           client -> replica in eu-west
              |                                      |
              | must confirm with a                  | answers from local
              | quorum spanning regions              | state immediately
              v                                      v
        us-east + ap-south                     response in 1 ms
              |
              v                              May be stale by the current
        response in ~120 ms                  replication lag (ms to s).

  Nothing is broken in either case. This is a design choice being paid for
  on EVERY request, not an exceptional condition.

The four PACELC classes and where real systems sit:

Class	During a partition	Normally	Examples
PC/EC	Consistency	Consistency	Spanner, etcd, ZooKeeper, HBase
PC/EL	Consistency	Latency	PNUTS
PA/EL	Availability	Latency	Dynamo, Cassandra, Riak (defaults)
PA/EC	Availability	Consistency	MongoDB (with majority writes)

# The same client library, three PACELC positions, chosen per query.
# Cassandra makes the trade explicit at the call site, which is the right
# design: the trade belongs to the operation, not to the deployment.

# PA/EL: fastest possible, may be stale. Right for a view counter.
session.execute(stmt, consistency_level=ConsistencyLevel.ONE)

# Stronger: R + W > N gives read-your-writes across the cluster at the cost
# of contacting a majority on every operation.
session.execute(stmt, consistency_level=ConsistencyLevel.QUORUM)

# PC/EC across regions: correct everywhere, and slow everywhere, because a
# quorum now spans continents and inherits the speed of light.
session.execute(stmt, consistency_level=ConsistencyLevel.EACH_QUORUM)

# Why the "E" half usually matters more than the "P" half, quantified.
PARTITION_MINUTES_PER_YEAR = 60          # a generous estimate for one system
MINUTES_PER_YEAR = 525_600

partition_fraction = PARTITION_MINUTES_PER_YEAR / MINUTES_PER_YEAR
print(f"partitioned: {partition_fraction:.4%} of the time")
# partitioned: 0.0114% of the time
#
# The CAP choice governs 0.01% of your operating life.
# The ELSE choice governs the other 99.99% -- and shows up in p99 latency on
# every dashboard you own. Teams routinely spend weeks debating the first and
# accept the second by accident, through a client-library default.

The practical value of PACELC is that it forces the second question to be asked out loud. "Are we CP or AP?" is a conversation about a rare event. "What are we paying in latency for consistency on the read path, and is that price worth it for this query?" is a conversation about every request the system serves.

Key Takeaways

PACELC extends CAP with the normal-operation trade: latency versus consistency when there is no partition.

Consistency requires coordination, and coordination costs a round trip — paid on every request, not only during incidents.

Systems are classified as PC/EC, PC/EL, PA/EL, or PA/EC; the best ones let you choose per operation.

The "else" branch governs virtually all of a system's lifetime and dominates observed latency.

🧪 Practice

Place these on the PACELC grid: PostgreSQL with synchronous replication, DynamoDB eventually-consistent reads, etcd, Cassandra with QUORUM.

For a three-region deployment, estimate the added read latency of QUORUM versus ONE and state which endpoints should pay it.

Interview: Your p99 read latency is 140 ms and the database is idle. What do you suspect? (Hint: an idle database that answers slowly is usually waiting for something other than disk.)

FLP Impossibility Result

The Fischer-Lynch-Paterson result of 1985 is the deepest constraint in the chapter. It states that in an asynchronous system where even one process may fail by crashing, there is no deterministic algorithm that guarantees consensus — agreement on a single value — in bounded time.

The word doing all the work is asynchronous, which here has a precise meaning: there is no bound on message delay and no bound on process speed. Under those conditions, a process that has not responded is indistinguishable from a process that is merely slow. And that single ambiguity is enough to defeat every deterministic protocol.

 THE AMBIGUITY AT THE HEART OF FLP

  node A waits for a message from node B.

    B crashed             -> A must proceed without B, or wait forever
    B is slow             -> A must wait, or it may violate agreement
    the message was slow  -> same as above

  In an asynchronous model these are INDISTINGUISHABLE, no matter how long
  A waits. Waiting longer never resolves it -- there is no bound to exceed.

  If A proceeds:  it may disagree with B (safety violated).
  If A waits:     it may wait forever (liveness violated).

  FLP: no deterministic algorithm escapes this. It is not about being clever.

The result is stated in terms of safety and liveness, and the distinction matters for everything that follows:

Safety: nothing bad happens — no two nodes decide different values.
Liveness: something good eventually happens — a decision is reached.

FLP says you cannot deterministically guarantee both under asynchrony with one possible crash. Every practical consensus system responds by keeping safety unconditionally and relaxing liveness: Raft and Paxos never allow two conflicting decisions, but they can stall indefinitely if the network behaves adversarially. In practice they make progress because real networks are not adversarial for long.

# What FLP forbids, and what real systems do instead.
def naive_consensus(nodes, value, timeout):
    """This function cannot be made correct. Included to show WHY."""
    responses = broadcast_and_wait(nodes, value, timeout)
    if len(responses) < majority(nodes):
        # We timed out. But a timeout does not mean the node is dead --
        # it means we stopped waiting. Choosing a timeout is choosing to
        # GUESS about failure, and the guess can be wrong in both directions.
        raise NoDecision()
    return decide(responses)

# The three real escapes from FLP -- none of which contradict it:
#
# 1. PARTIAL SYNCHRONY (Raft, Paxos, Zab): assume the network is EVENTUALLY
#    well-behaved. Safety always holds; liveness holds once messages start
#    arriving within some bound. Timeouts become failure DETECTORS, and a
#    wrong guess costs an extra election, never a wrong decision.
#
# 2. RANDOMIZATION (Ben-Or, and modern BFT protocols): terminate with
#    probability 1 rather than with certainty. FLP applies to deterministic
#    algorithms, so a coin flip sidesteps it.
#
# 3. FAILURE DETECTORS (Chandra-Toueg): assume an oracle that eventually
#    stops suspecting correct processes. This is a formalization of what a
#    well-tuned heartbeat gives you in practice.

The practical consequence appears everywhere in this chapter: every timeout in a distributed system is a guess about whether a remote process is dead, and because FLP guarantees the guess can be wrong, systems must be correct even when it is. That is why Raft uses randomized election timeouts (to avoid repeated split votes), why leases have fencing tokens (so a wrongly-suspected leader cannot corrupt state when it returns), and why a "dead" node reappearing must never be able to act as if no time has passed.

Response to FLP	What it gives up	Used by
Partial synchrony	Guaranteed termination time	Raft, Paxos, Zab, Viewstamped Replication
Randomization	Determinism	Ben-Or, HoneyBadgerBFT
Failure detectors	Perfect accuracy	Chandra-Toueg family
Abandon consensus	Agreement itself	CRDTs, gossip, AP stores

Key Takeaways

FLP: in an asynchronous system with one possible crash failure, no deterministic algorithm guarantees consensus in bounded time.

The obstacle is that a crashed process and a slow process are indistinguishable without a bound on delay.

Real protocols keep safety unconditionally and accept that liveness may stall, assuming partial synchrony to make progress in practice.

Every timeout is a guess about failure that can be wrong, so correctness must never depend on the guess being right.

🧪 Practice

Explain why waiting longer never resolves the crashed-versus-slow ambiguity in the asynchronous model.

For Raft, name which property is guaranteed unconditionally and which is conditional, and on what.

Interview: If consensus is impossible, how does etcd work? (Hint: the theorem's assumptions include one that real networks violate in a helpful direction.)

Partial Failure

In a single-process program, failure is total: the process either runs or it crashes, and every part of it shares that fate. In a distributed system, failure is partial — some components fail while others continue, and the survivors often cannot tell which is which.

This is the qualitative difference that makes distributed systems hard. A local function call has two outcomes: it returns or it throws. A remote call has at least five, and one of them has no observable difference from another.

 LOCAL CALL                    REMOTE CALL

  result = f(x)                 result = svc.f(x)

  outcomes:                     outcomes:
   1. returns a value            1. executed, response received
   2. raises                     2. executed, response LOST     -> looks like 4
                                 3. never arrived               -> looks like 4
                                 4. timed out (unknown)
                                 5. partially executed, then the callee died
                                    (the write landed; the event was never sent)

  Outcome 5 is the dangerous one: the system is now in a state that no
  single component's code believed was possible.

Partial failure produces a specific family of bugs that do not exist in single-process software:

Inconsistent state across components. A handler writes to the database, then crashes before publishing the event. The database says the order exists; the downstream services never heard of it. Neither is wrong; the system is inconsistent because the two writes were not atomic. This is precisely the problem the outbox pattern (Section 8.5) exists to solve.

Gray failure. A node that is neither up nor down — responding to health checks while failing real requests, or serving at 5% of normal speed. These are harder than crashes, because every automated mechanism keyed on "is it responding?" concludes that it is fine.

Cascading failure. One component's slowness consumes the callers' threads, which makes the callers slow, which consumes their callers' threads. The original failure is small; the outage is total.

# The classic partial-failure bug, and why it is not fixable with try/except.
def place_order(order):
    order_id = db.insert_order(order)        # succeeds
    # X  process killed here (OOM, deploy, node failure)
    events.publish("OrderPlaced", order_id)  # never runs
    return order_id
# Result: an order exists that no downstream service will ever process. No
# exception was raised anywhere; no log line records a failure. The order is
# simply invisible to the rest of the company.

# The fix is not error handling -- it is making the two writes atomic by
# putting them in the same transactional boundary (the outbox pattern).
def place_order(order):
    with db.transaction():
        order_id = db.insert_order(order)
        db.insert_outbox("OrderPlaced", {"order_id": order_id})   # SAME txn
    return order_id
# A separate relay publishes from the outbox table with at-least-once
# delivery. A crash anywhere now leaves a consistent state: either both rows
# exist or neither does.

Designing for partial failure means assuming that every remote interaction can half-happen, and structuring the system so that a half-happened operation is either completed or undone rather than left ambiguous:

Technique	What it addresses
Idempotency	Safe retries after ambiguous outcomes
Timeouts on every call	Prevents unbounded waiting on a dead peer
Circuit breakers	Stops calls to a failing dependency
Bulkheads	Confines one failure to its own resource pool
Atomic write boundaries	Removes the "wrote one, not the other" state
Reconciliation jobs	Detects and repairs drift that slipped through

The last one is underused and deserves emphasis. However careful the design, some inconsistency will occur; a periodic job that compares two sources of truth and reports or repairs divergence is the only mechanism that finds problems nobody anticipated. Systems that handle money almost always have one.

Key Takeaways

Partial failure means some components fail while others continue, and survivors often cannot determine which.

A remote call has at least five outcomes, several of which are indistinguishable from a timeout.

The characteristic bug is a half-completed multi-step operation; the fix is an atomic write boundary, not better error handling.

Gray failures defeat health checks, and reconciliation jobs are what catch the drift no design anticipated.

🧪 Practice

List every partial-failure state of "charge the card, create the order, send the email" and say which are recoverable.

Explain why a try/finally around the publish call does not fix the order bug above.

Interview: How would you detect that your database and search index have silently drifted apart? (Hint: what would you compare, and how often, given that neither reported an error?)

Network Partitions

A network partition is a failure that splits a system into groups of nodes that cannot communicate with each other, while nodes within each group communicate normally. It is the specific condition that CAP is about, and it is far more varied — and more common — than the textbook picture of a cleanly severed cable suggests.

The reason partitions are treated as a category of their own is that, unlike a crash, the isolated nodes keep running. A crashed node does nothing. A partitioned node continues to serve requests, accept writes, and believe it is healthy, all while operating on a view of the world that is diverging from the other side's. Two halves of a system can both be "up" and both be wrong.

 PARTITION TYPES (increasing order of difficulty)

 1. COMPLETE            2. PARTIAL              3. ASYMMETRIC (one-way)

   A B | C D              A --- B                 A ---> B    (A reaches B)
       |                  |  \  |                 A <-X- B    (B cannot reply)
   two groups, no         |   X |
   cross-traffic          C --- D                 Each node's view of who is
                                                  reachable DISAGREES with the
   Easiest: quorum        C cannot reach B,       other's -- failure detectors
   resolves it            everyone else can       give contradictory answers

 4. INTERMITTENT (flapping)         5. SLOW (a partition in effect)

   connectivity returns and          messages arrive after 30 seconds. Nothing
   drops every few seconds           is "down"; every timeout fires anyway.
   -> repeated elections,            -> the system behaves as partitioned while
      repeated failovers                every health check reports success

The most dangerous consequence is split brain: both sides elect a leader, both accept writes, and the two histories diverge irreconcilably. When the partition heals, there is no correct merge — two customers were sold the last item, two conflicting balances exist, and someone must decide which reality to discard.

Quorum is the standard defence. Requiring a majority to make progress guarantees that at most one side can have one, since two majorities of the same set must overlap:

def can_make_progress(reachable_nodes, cluster_size):
    """Majority quorum: at most one partition can satisfy this, ever."""
    return reachable_nodes > cluster_size // 2

# 5-node cluster split 3/2:
#   side of 3 -> 3 > 2 -> proceeds as the authoritative side
#   side of 2 -> 2 > 2 is False -> refuses writes, steps down if it was leader
#
# Note what this costs: the minority side is UNAVAILABLE even though its nodes
# are perfectly healthy. That is the CAP choice being made mechanically.

def tolerated_failures(cluster_size):
    return (cluster_size - 1) // 2

for n in (1, 2, 3, 4, 5, 6, 7):
    print(f"{n} nodes -> tolerates {tolerated_failures(n)} failures")
# 3 nodes -> 1     5 nodes -> 2     7 nodes -> 3
# 4 nodes -> 1     6 nodes -> 2
#
# EVEN CLUSTER SIZES ARE WASTEFUL: 4 nodes tolerate the same single failure
# as 3, while adding a node that must be paid for and coordinated with.
# This is why quorum systems are almost always deployed with odd sizes.

Quorum alone is not sufficient, because a leader that is partitioned away does not instantly know it has lost quorum. For a brief window it may still believe it is the leader and act on that belief. Two mechanisms close the gap:

Leases with expiry. A leader holds a time-bounded lease and must stop acting before it expires unless renewed. Combined with a follower rule of "do not elect a new leader until the old lease could not still be valid", this bounds the overlap.

Fencing tokens. Every leadership term gets a monotonically increasing number, and downstream systems reject writes carrying a token lower than the highest they have seen. A stale leader that returns after a partition is rejected by the storage layer itself, regardless of what it believes.

# Fencing: the storage layer enforces what the coordinator cannot guarantee.
def write_with_fence(resource, value, token):
    current = store.get_fence_token(resource)
    if token < current:
        # A previously-partitioned leader has come back and is still acting.
        # It cannot be trusted to know it was replaced -- reject it here.
        raise StaleLeader(f"token {token} < {current}")
    store.put(resource, value, fence_token=token)

A final point on testing. Partitions are rare enough that they are never naturally exercised before production, which is why deliberate fault injection — blocking traffic between nodes, adding one-way latency, flapping links — is the only way to know how a system behaves in the condition its entire design is premised on.

Key Takeaways

A partition leaves nodes running but unable to communicate, so both sides can be simultaneously healthy and divergent.

Partitions are frequently partial, asymmetric, flapping, or expressed as extreme slowness rather than clean splits.

Majority quorum guarantees at most one side can proceed, at the cost of the minority's availability; use odd cluster sizes.

Quorum needs leases and fencing tokens to handle a stale leader that has not yet learned it was replaced.

🧪 Practice

For a 6-node cluster, compute the failures tolerated and explain why 5 or 7 would be a better choice.

Describe a concrete split-brain scenario in an inventory system and the business consequence when the partition heals.

Interview: Your leader is partitioned from the followers but can still reach the database. What breaks, and which mechanism prevents it? (Hint: the leader does not yet know it has been replaced — who else could know?)

8.2 Consistency Models

A consistency model is a contract between a data store and its users: given these operations, which results are legal? The models form a hierarchy from strongest and slowest to weakest and fastest, and choosing among them is one of the highest-leverage decisions in a distributed design.

Strong Consistency and Linearizability

Linearizability is the strongest single-object consistency model. It guarantees that the system behaves as if there were exactly one copy of the data and every operation took effect instantaneously at some point between when it was invoked and when it returned. Operations appear in a single total order, and that order respects real time: if operation A completes before operation B begins, A comes first.

The reason this is worth wanting is that it makes a distributed system indistinguishable from a single machine. Every subtle question — can I read a value I just wrote, can two clients see different orders of the same events, can a value go backwards — has the obvious answer. Linearizability is the model that requires no reasoning at all, which is why it is the right default whenever it is affordable.

The analogy is a single physical ledger on one desk. Everyone writes and reads from that one book, one at a time. Nobody can see a page that was already overwritten, and if you finish writing before I start reading, I see your entry.

 LINEARIZABLE (legal)                 NOT LINEARIZABLE (illegal)

 client A: |--write(x=1)--|           client A: |--write(x=1)--|
 client B:                 |-read->1  client B:                 |-read->0
 client C:                  |-read->1 client C:                  |-read->1
                                                                    ^
 Once the write COMPLETES, every                B read 0 AFTER the write
 later read sees it. There is one               completed, then C read 1.
 point in time where x becomes 1.               The value appears to go
                                                backwards in real time.

Linearizability is a recency guarantee about single objects, and both halves of that phrase matter. It says nothing about multi-object operations — for that you need serializability, which is a transactional property. A system can be linearizable per key and still allow you to observe a state where money left one account and has not yet arrived in the other.

# Achieving linearizability costs coordination on BOTH paths.
# Reading from a leader is not sufficient: the leader may have been deposed
# by a partition and not yet know it.

def linearizable_read(key):
    # Option 1: ReadIndex -- confirm we are still leader by exchanging a
    # heartbeat round with a quorum before answering.
    if not raft.confirm_still_leader_via_quorum():   # one round trip
        raise NotLeader()
    return raft.state_machine.get(key)

def lease_read(key):
    # Option 2: leader leases -- cheaper, but its correctness depends on a
    # bound on clock drift, which is an assumption about hardware rather
    # than about messages.
    if time.monotonic() > raft.lease_expiry:
        return linearizable_read(key)                # fall back to a quorum
    return raft.state_machine.get(key)               # no network round trip

def stale_read(key):
    # NOT linearizable: fast, and may return a value that was already
    # superseded. Legitimate for many reads -- but say so explicitly.
    return local_replica.get(key)

The costs are unavoidable and follow directly from the definition. Every linearizable operation requires coordination, so latency is bounded below by a round trip to a quorum — across regions, that is tens to hundreds of milliseconds. Availability suffers because a minority partition must refuse service. And throughput is bounded by the coordination path, not by the storage.

System	Linearizable?
etcd, ZooKeeper, Consul	Yes, by design
Spanner	Yes (TrueTime bounds clock uncertainty)
PostgreSQL single primary	Yes for reads on the primary
PostgreSQL async replicas	No — replicas lag
DynamoDB	Optional per read (`ConsistentRead`)
Cassandra `QUORUM`	Close, but not strictly linearizable
Redis with replicas	No — asynchronous replication

Use linearizability where a stale answer causes an incorrect decision: distributed locks, leader election, unique constraint enforcement, inventory decrements, account balances. Avoid paying for it on timelines, counters, recommendations, and search results, where nobody can perceive the difference.

Key Takeaways

Linearizability makes a distributed system behave like a single copy with operations ordered in real time.

It is a per-object recency guarantee, not a multi-object transactional one.

It requires coordination on reads as well as writes; leader reads alone are not sufficient without a quorum check or a lease.

Reserve it for operations where staleness causes a wrong decision, since it costs a quorum round trip on every access.

🧪 Practice

Draw a timeline of three clients that is sequentially consistent but not linearizable.

Explain why reading from a Raft leader without a ReadIndex or lease can return stale data.

Interview: Why is linearizability insufficient for a bank transfer? (Hint: count how many objects the operation touches.)

Sequential Consistency

Sequential consistency, defined by Leslie Lamport in 1979, requires that all operations appear in some single total order, and that each individual process's operations appear in that order in the sequence it issued them. What it drops, relative to linearizability, is the requirement that the order respect real time.

That single omission is the whole difference. Under linearizability, if your write finishes at 10:00:00 and my read starts at 10:00:01, I must see your write. Under sequential consistency, I need not — as long as everyone agrees on some consistent order, and nobody sees any single process's own operations out of order.

The analogy is a group chat where every participant sees the same final transcript, but messages may appear in a different order than they were actually sent — as long as each person's own messages stay in their sent order, and everybody's transcript is identical.

 SEQUENTIALLY CONSISTENT BUT NOT LINEARIZABLE

  real time -->
  P1: |--write(x=1)--|
  P2:                    |--write(y=1)--|
  P3:                                     |-read(y)->1--| |-read(x)->0--|

  P3 sees y=1 but x=0, even though x was written FIRST in real time.

  Legal under sequential consistency: everyone can agree on the order
    write(y=1), read(y)->1, read(x)->0, write(x=1)
  which is internally consistent and preserves each process's own order.

  Illegal under linearizability: write(x=1) completed before read(x) began,
  so the read must return 1.

The practical significance is that sequential consistency permits an implementation to serve reads from a replica that is behind, provided the replica applies updates in a consistent global order. That removes the real-time coordination requirement and can substantially reduce read latency, while still giving applications a single coherent view of history.

# A sequentially consistent replica: applies a globally ordered log, but may
# be behind. All replicas converge to the SAME order -- just not at the same
# real time.
class SequentialReplica:
    def __init__(self):
        self.state, self.applied_seq = {}, 0

    def apply(self, seq, key, value):
        # Operations arrive with a globally assigned sequence number and are
        # applied strictly in order. Gaps are waited for, never skipped --
        # skipping is what would break the total order.
        assert seq == self.applied_seq + 1, "out-of-order application"
        self.state[key] = value
        self.applied_seq = seq

    def read(self, key):
        # May return a value that is older than the true current one. What it
        # can NEVER do is return a state that no ordering would produce --
        # for example, an effect without its cause.
        return self.state.get(key)

Where you meet sequential consistency in practice is less often a database setting than a hardware and language-runtime concept: it is the memory model that CPUs and compilers approximate, and Java's volatile and C++'s memory_order_seq_cst provide it for individual variables. In distributed data stores it appears mostly as an intermediate rung — stronger than causal, weaker than linearizable — and relatively few systems offer it as an explicit option, because the cost saving relative to linearizability is smaller than the cost saving of dropping all the way to causal or eventual.

Model	Total order	Respects real time	Per-process order
Linearizable	Yes	Yes	Yes
Sequential	Yes	No	Yes
Causal	No	No	Yes
Eventual	No	No	Not guaranteed

The composability difference is worth knowing for interviews and for design reviews: linearizability composes, sequential consistency does not. If every object in a system is individually linearizable, the system as a whole is linearizable. If every object is individually sequentially consistent, the combination may not be — the orders chosen per object need not agree with one another.

Key Takeaways

Sequential consistency requires one agreed total order that preserves each process's own operation order, but need not match real time.

It permits reads from replicas that lag, provided all replicas apply the same order.

It is most familiar as a memory model (volatile, seq_cst) rather than a database setting.

Unlike linearizability, it does not compose across objects.

🧪 Practice

Give an execution that is sequentially consistent, and explain which linearizability requirement it violates.

Explain why the replica class above must wait for a gap rather than skip it.

Interview: Two objects are each sequentially consistent. Is the pair? (Hint: each object gets to pick its own order — must those two choices agree?)

Causal Consistency

Causal consistency guarantees that operations related by cause and effect are seen in the same order by everyone, while operations that are unrelated (concurrent) may be seen in different orders by different observers.

The motivation is that most anomalies users actually notice are causality violations, not ordering violations in the abstract. Seeing a reply before the message it replies to is confusing. Seeing two unrelated posts in a different order than your friend did is invisible. Causal consistency spends coordination only where it buys something perceptible.

 THE ANOMALY CAUSAL CONSISTENCY PREVENTS

  Alice posts:      "I lost my job"          (event 1)
  Alice deletes it, posts: "April fools!"    (event 2, CAUSED BY event 1)
  Bob replies:      "That's terrible!"       (event 3, CAUSED BY event 1)

  Under EVENTUAL consistency, Carol may see:
      "That's terrible!"        <- a reply
      "I lost my job"           <- the thing replied to, arriving after
  or worse, Bob's reply attached to a post that Carol never received.

  Under CAUSAL consistency, Carol is guaranteed to see event 1 before
  event 3, because event 3 causally depends on event 1. Whether she sees
  event 2 before or after some UNRELATED post by Dave is left open --
  and nobody can tell the difference.

Causality is established by the happens-before relation (covered fully in Section 8.3): operation A happens-before B if A occurred earlier in the same process, or if B read a value that A wrote, or by transitivity through a chain of such links. Everything not related by happens-before is concurrent, and concurrent operations may be ordered arbitrarily.

# Causal consistency is implemented by shipping dependencies with each write.
class CausalStore:
    def __init__(self, node_id):
        self.node_id = node_id
        self.state = {}
        self.clock = {}        # vector clock: node -> highest seen counter
        self.pending = []      # writes whose dependencies are not yet met

    def write(self, key, value):
        self.clock[self.node_id] = self.clock.get(self.node_id, 0) + 1
        # The write carries the writer's full causal history. Anyone applying
        # it must first have applied everything the writer had seen.
        return {"key": key, "value": value, "deps": dict(self.clock),
                "origin": self.node_id}

    def read(self, key):
        value = self.state.get(key)
        # Reading creates a causal dependency: anything this process writes
        # next is causally after the write it just observed.
        if key in self.write_clocks:
            self._merge(self.write_clocks[key])
        return value

    def receive(self, update):
        if self._deps_satisfied(update["deps"]):
            self._apply(update)
            self._drain_pending()          # applying one may unblock others
        else:
            self.pending.append(update)    # BUFFER: do not apply an effect
                                           # before its cause has arrived

    def _deps_satisfied(self, deps):
        # Every dependency the writer had must already be reflected locally.
        return all(self.clock.get(n, 0) >= c
                   for n, c in deps.items() if n != update_origin(deps))

Causal consistency sits at an important theoretical boundary: it is the strongest model that remains available during a network partition. Both sides of a partition can continue accepting reads and writes, buffering updates until dependencies arrive, and converge afterwards without ever showing an effect before its cause. Nothing stronger has that property, which is why it is the target for AP systems that still want to avoid visible nonsense.

Model	Available under partition	Prevents causality violations
Linearizable	No	Yes
Sequential	No	Yes
Causal	Yes	Yes
Eventual	Yes	No

The cost is metadata. Tracking causality precisely requires vector clocks or dependency lists that grow with the number of writers, which is why production systems approximate: they track dependencies per session, or per partition, or truncate old entries. That approximation is usually acceptable, because the causal chains users can perceive are short.

Key Takeaways

Causal consistency orders causally related operations identically for all observers and leaves concurrent operations unordered.

It prevents the anomalies users actually notice — effects appearing before their causes — without global coordination.

It is the strongest consistency model that remains available during a partition.

Its cost is causal metadata that grows with the number of writers, so real systems approximate it per session or partition.

🧪 Practice

Give a three-message chat sequence where eventual consistency produces a visible anomaly and causal consistency does not.

Explain why the receive method buffers instead of applying immediately, and what would break otherwise.

Interview: Why can causal consistency stay available during a partition when sequential consistency cannot? (Hint: what would the two sides have to agree on, and can they do it without talking?)

Eventual Consistency

Eventual consistency makes one promise: if writes stop, all replicas will eventually converge to the same value. It says nothing about how long that takes, what you read in the meantime, or whether successive reads move forward or backward.

Stated that plainly it sounds almost useless, and the honest response is that as a guarantee it is very weak. Its value is not in what it promises but in what it permits: because no coordination is required before answering, every replica can serve reads and accept writes locally, at local latency, during partitions, in any region. For the enormous class of data where a few seconds of staleness is imperceptible, that is an excellent trade.

The analogy is a rumour spreading through an office. Nobody coordinates; each person tells the people near them. For a while different parts of the office believe different things. Given a pause in new gossip, everyone converges on the same story.

 CONVERGENCE OVER TIME

  t=0   write x=5 at replica A
        A: x=5    B: x=(old)   C: x=(old)     <- divergent, all readable
  t=1   A gossips to B
        A: x=5    B: x=5       C: x=(old)
  t=2   B gossips to C
        A: x=5    B: x=5       C: x=5         <- converged

  Between t=0 and t=2, a client reading from C gets the old value while a
  client reading from A gets the new one. Both replicas are healthy and
  correct according to the model.

The critical engineering question is not whether replicas converge but what they converge to when two replicas accepted conflicting writes. That is a conflict-resolution decision, and it is where eventual consistency stops being free:

Strategy	Behaviour	Data loss risk
Last-write-wins	Highest timestamp survives	Silently drops writes
Highest version	Explicit version counter wins	Same, but explicit
Application merge	Custom logic reconciles both	None if written well
CRDTs	Mathematically guaranteed convergence	None, by construction
Sibling values	Both retained; client resolves	None; complexity moves to the client

# The anti-guarantee that surprises people: reads can go BACKWARDS.
# Nothing in eventual consistency forbids this.
v1 = read_from(replica_a)     # 42  (has the recent write)
v2 = read_from(replica_b)     # 17  (does not yet)
v3 = read_from(replica_a)     # 42
# A user refreshing a page sees 42, 17, 42. Every read is "correct".
#
# The fix is not a stronger global model -- it is a SESSION guarantee
# (next topic): pin the session to one replica, or track the highest version
# the session has observed and never serve it something older.

# Measuring what eventual actually means for you -- the number that decides
# whether the model is acceptable for a given feature.
def replication_lag_budget(p99_lag_ms, user_action_gap_ms):
    """A staleness window is only acceptable if users cannot outrun it."""
    return {
        "p99_lag_ms": p99_lag_ms,
        "user_returns_within_ms": user_action_gap_ms,
        "anomaly_visible": user_action_gap_ms < p99_lag_ms,
    }

print(replication_lag_budget(p99_lag_ms=250, user_action_gap_ms=100))
# {'p99_lag_ms': 250, 'user_returns_within_ms': 100, 'anomaly_visible': True}
#
# A user who posts a comment and re-renders the page in 100 ms WILL see their
# own comment missing. "Eventually consistent" is fine for other people's
# data and unacceptable for your own -- hence read-your-writes.

Eventual consistency is right for view counts, recommendation results, search indexes, social feeds, analytics aggregates, and DNS. It is wrong wherever a stale read causes an irreversible decision: inventory decrements, balance checks, permission and revocation checks, and uniqueness enforcement.

Key Takeaways

Eventual consistency promises only convergence in the absence of new writes — no bound on staleness and no ordering guarantees.

Its value is permitting local reads and writes with no coordination, which makes it fast and partition-tolerant.

Reads can move backwards; session guarantees, not stronger global models, are the usual fix.

The real design work is conflict resolution: last-write-wins silently loses data, while CRDTs converge without loss.

🧪 Practice

Name four data types where eventual consistency is clearly right and three where it is clearly wrong. State the deciding question.

Explain how a user can observe a value moving backwards, and give two fixes.

Interview: A product manager asks "how eventual is eventually?". What do you measure and report? (Hint: it is a latency distribution, not a constant.)

Read-Your-Writes and Monotonic Reads

Global consistency models describe what all clients see collectively. Session guarantees describe what a single client sees over the course of its own session — and they turn out to fix most of the anomalies users actually complain about, at a fraction of the cost of strengthening the global model.

The insight is that a user does not perceive global inconsistency. They perceive their own inconsistency: their comment vanishing, a number going backwards, their profile update not appearing. Four session guarantees address exactly these:

Guarantee	Promise
Read-your-writes	You always see your own writes
Monotonic reads	You never see the data move backwards
Monotonic writes	Your writes are applied in the order you issued them
Writes-follow-reads	Your write is ordered after the write you read before it

Read-your-writes (read-after-write consistency) is the most important. Its absence produces the single most-reported bug in replicated systems: the user edits their profile, the page reloads from a lagging replica, and the change is gone. They edit again. Now there are two writes, and support has a ticket.

 WITHOUT READ-YOUR-WRITES               WITH READ-YOUR-WRITES

  user --write--> [ primary ]            user --write--> [ primary ]
                       | async                                | records the
  user --read---> [ replica ]            user --read--> routed to primary
                    (behind)                            (or to a replica
                       |                                 known to be caught up)
                 sees the OLD value
                 "my edit didn't save!"                 sees the NEW value

There are four implementations, in increasing order of sophistication:

# 1. Route to the primary for a window after any write. Simple, effective,
#    and it concentrates load on the primary -- fine when writes are rare.
def read(user_id, key):
    if session.last_write_at and time.time() - session.last_write_at < 5.0:
        return primary.get(key)
    return replica.get(key)

# 2. Sticky routing: pin the session to one replica. Gives monotonic reads
#    too, but breaks when that replica is removed or rebalanced.

# 3. Version tokens -- the most precise and the most portable. The client
#    carries the version it has observed; the server refuses to serve older.
def write(key, value):
    version = primary.put(key, value)
    session.max_version = max(session.max_version, version)   # remember it
    return version

def read(key):
    replica = pick_replica()
    if replica.applied_version < session.max_version:
        # This replica has not caught up to what this session already saw.
        replica = primary                    # or wait, or try another replica
    return replica.get(key)

# 4. Write-through cache: serve the session its own write from a local or
#    edge cache while replication catches up. Common in web frameworks.

Monotonic reads prevents the mirror-image anomaly: values moving backwards because successive reads landed on replicas with different lag. The fix is the same version-token mechanism — never serve a session a version lower than the highest it has already observed.

# Monotonic reads via a high-water mark carried in the session.
def monotonic_read(key):
    seen = session.versions.get(key, 0)
    for replica in candidate_replicas():
        value, version = replica.get_with_version(key)
        if version >= seen:                 # never go backwards for THIS session
            session.versions[key] = version
            return value
    return primary.get_with_version(key)[0]  # last resort: the source of truth

The reason session guarantees are such good value is arithmetic. Making the whole system linearizable means every operation pays a quorum round trip. Providing read-your-writes means a small fraction of reads — those from sessions that recently wrote — are routed differently. In a typical read-heavy workload that is a few percent of traffic paying the cost, instead of all of it.

# Why the cheap fix is usually the right one.
READ_QPS, WRITE_QPS, WINDOW_S = 50_000, 500, 5.0
sessions_needing_primary = WRITE_QPS * WINDOW_S          # 2,500 concurrent
affected_reads = sessions_needing_primary * 0.4          # ~1,000 reads/s

print(f"{affected_reads / READ_QPS:.1%} of reads need the primary")
# 2.0% of reads need the primary
#
# Two percent of reads pay for correctness, and the user-visible anomaly is
# gone. Full linearizability would have charged 100% of reads for it.

Key Takeaways

Session guarantees fix the anomalies a single user perceives, which is where nearly all complaints originate.

Read-your-writes prevents "my edit disappeared"; monotonic reads prevents values moving backwards.

Version tokens carried in the session are the most precise and portable implementation; primary routing is the simplest.

They cost a small fraction of traffic, versus strengthening the global model which charges every operation.

🧪 Practice

Implement read-your-writes using version tokens for a system with one primary and five replicas, handling the case where no replica has caught up.

Explain how sticky routing accidentally provides three of the four session guarantees, and which one it misses.

Interview: A user updates their avatar and it reverts on refresh, then appears again. Which guarantees are missing? (Hint: two distinct anomalies are described — identify each one separately.)

Tunable Consistency

Tunable consistency lets the caller choose the consistency level per operation rather than accepting one setting for the whole system. It is the practical answer to everything above: since different operations have different correctness requirements, forcing them all to the strongest one is wasteful and forcing them all to the weakest one is wrong.

The mechanism, used by Dynamo-style systems, is quorum arithmetic. With N replicas, a write is acknowledged after W of them confirm and a read consults R of them. The relationship that matters is:

If R + W > N, the read set and the write set must overlap, so any read is guaranteed to see at least one replica holding the latest acknowledged write.

 QUORUM OVERLAP, N = 3

  W=1, R=1  (R+W=2, NOT > 3)        W=2, R=2  (R+W=4 > 3)

   write -> [A]  B   C               write -> [A][B]  C
   read  ->  A   B  [C]              read  ->     [B][C]
                    ^                              ^
   No overlap: the read may           Overlap guaranteed: B holds the
   miss the write entirely.           write, so the read sees it.

   Fast, weak.                        Slower, strong enough for
                                      read-your-writes cluster-wide.

# The arithmetic, made concrete.
def quorum_properties(N, R, W):
    return {
        "strong_read": R + W > N,          # read and write sets intersect
        "write_survives": W > N // 2,      # survives a minority partition
        "node_failures_tolerated_read": N - R,
        "node_failures_tolerated_write": N - W,
    }

for R, W in [(1, 1), (1, 3), (2, 2), (3, 1)]:
    print(f"N=3 R={R} W={W}: {quorum_properties(3, R, W)}")

# N=3 R=1 W=1: strong_read=False -- fastest both ways, no guarantee
# N=3 R=1 W=3: strong_read=True  -- fast reads, writes fail if ANY node is down
# N=3 R=2 W=2: strong_read=True  -- balanced; the usual default
# N=3 R=3 W=1: strong_read=True  -- fast writes, reads fail if any node is down
#
# The choice is about which side you want to be fast and which you can afford
# to have fail: R=1,W=3 optimizes reads and makes writes fragile, and vice
# versa. R=W=2 is the compromise almost everyone lands on.

# Per-operation levels in practice (Cassandra syntax; DynamoDB and others
# expose the same idea with fewer knobs).
from cassandra import ConsistencyLevel as CL

# Analytics read: staleness is irrelevant, latency is everything.
session.execute(view_count_query, consistency_level=CL.ONE)

# Inventory decrement: must not oversell.
session.execute(decrement_stock, consistency_level=CL.QUORUM)

# A genuine compare-and-set needs more than quorum -- it needs consensus,
# because two concurrent quorum writes can both succeed.
session.execute("""
    UPDATE inventory SET qty = 4 WHERE sku = 'X' IF qty = 5
""")   # lightweight transaction: Paxos under the hood, ~4x the latency

# Multi-region correctness at multi-region cost.
session.execute(critical_write, consistency_level=CL.EACH_QUORUM)

Level	Meaning	Typical use
`ONE`	One replica responds	Metrics, logs, view counts
`QUORUM`	Majority of all replicas	The general-purpose default
`LOCAL_QUORUM`	Majority within the local datacentre	Multi-region, latency-aware
`EACH_QUORUM`	Majority in every datacentre	Cross-region correctness
`ALL`	Every replica	Rarely — one node down fails everything

Two cautions temper the flexibility.

Quorum is not consensus. R + W > N guarantees a read sees the latest acknowledged write; it does not serialize concurrent writes. Two clients doing read-modify-write at QUORUM can still lose one update. Anything conditional needs a compare-and-set backed by real consensus, which costs several times more.

LOCAL_QUORUM is the multi-region default for a reason. It keeps latency local while retaining a majority within the region, accepting cross-region staleness. Choosing QUORUM in a three-region cluster silently makes every operation pay a cross-continent round trip.

Key Takeaways

Tunable consistency sets the level per operation, matching cost to the actual correctness requirement.

R + W > N guarantees the read and write sets overlap, so reads observe the latest acknowledged write.

Lowering R speeds reads and makes writes fragile, and vice versa; R = W = ceil((N+1)/2) is the usual balance.

Quorum is not consensus — read-modify-write still needs compare-and-set, and LOCAL_QUORUM is what keeps multi-region latency sane.

🧪 Practice

For N = 5, list every (R, W) pair giving strong reads and rank them by read latency and write availability.

Show how two concurrent QUORUM read-modify-writes can lose an update despite R + W > N.

Interview: A three-region cluster has p99 writes of 180 ms and the application is in one region. What do you check? (Hint: which quorum is the client asking for, and how far does that quorum reach?)

8.3 Time and Ordering

Ordering events is trivial on one machine — there is one clock and one sequence. Across machines there is no shared clock, and the mechanisms that substitute for one determine what a system can know about what happened before what.

Physical Clocks and Clock Skew

Every machine has a physical clock, and the natural instinct is to timestamp events with it and sort. That instinct produces a specific class of bug that is hard to reproduce and easy to misdiagnose, because the clocks involved are usually right — just not right enough, and not right at the same moment.

A computer's clock is a quartz oscillator counting ticks. Quartz frequency varies with temperature, voltage, and manufacturing tolerance, so every clock drifts. Typical drift is 10-100 parts per million, which sounds negligible until it is converted into a duration:

# What "a good clock" means in practice.
DRIFT_PPM = 50                       # a typical, well-behaved server clock

for hours in (1, 24, 24 * 7, 24 * 30):
    drift_ms = DRIFT_PPM * 1e-6 * hours * 3600 * 1000
    print(f"{hours:>4}h uncorrected -> {drift_ms:>8.1f} ms of drift")

#    1h uncorrected ->    180.0 ms
#   24h uncorrected ->   4320.0 ms
#  168h uncorrected ->  30240.0 ms
#  720h uncorrected -> 129600.0 ms   (over two minutes)
#
# Two servers drifting in OPPOSITE directions are twice as far apart. Without
# synchronization, timestamps from different machines are not comparable at
# any precision that matters for ordering.

Clock skew is the difference between two clocks at the same instant. Even under active synchronization, skew of 1-100 ms between machines in a datacentre is normal, and across regions or in virtualized environments it is worse. VMs are particularly bad: a paused or migrated VM can resume with its clock arbitrarily behind.

The failure this causes is silent data loss under last-write-wins:

 LAST-WRITE-WINS WITH SKEWED CLOCKS

  Node A clock: 10:00:00.500 (running 300 ms FAST)
  Node B clock: 10:00:00.100 (running 100 ms slow)

  real 10:00:00.200  Alice writes name="Alice" on node A -> ts 10:00:00.500
  real 10:00:00.400  Bob   writes name="Bob"   on node B -> ts 10:00:00.100
                                                              ^
  Bob's write happened LATER in reality but carries an EARLIER timestamp.
  Last-write-wins keeps Alice's. Bob's write is silently discarded, with no
  error, no conflict, and no log line. Bob sees his change "not save".

The essential distinction every engineer should internalize is between the two clocks a machine offers:

Clock	Measures	Jumps?	Use for
Wall clock	Time of day (`time.time()`)	Yes — NTP steps, leap seconds, manual changes	Timestamps for humans, logs, expiry dates
Monotonic	Elapsed time since some boot-relative origin	No, never goes backwards	Durations, timeouts, rate limiting

import time

# WRONG: wall clock for a duration. An NTP correction mid-measurement can
# make this negative, or inflate it by the size of the step.
start = time.time()
do_work()
elapsed = time.time() - start          # may be negative; has caused real
                                       # outages when used for timeouts

# RIGHT: monotonic clock for durations. Guaranteed non-decreasing.
start = time.monotonic()
do_work()
elapsed = time.monotonic() - start     # always sane

# Wall clock is correct for what it is FOR: recording when something
# happened for a human to read, or an absolute expiry.
record = {"event": "login", "at": time.time()}

The rule that follows is blunt and worth memorizing: never use wall-clock timestamps to order events across machines, and never use them to measure elapsed time. They are for display and for absolute deadlines. Ordering requires the logical clocks covered in the next several topics, or a system like Spanner's TrueTime that makes the uncertainty explicit and waits it out.

Key Takeaways

Quartz drift of 10-100 ppm means uncorrected clocks diverge by seconds per day and minutes per month.

Clock skew between synchronized machines is still milliseconds, and far worse on VMs that pause or migrate.

Wall clocks jump — forwards and backwards — so they must never be used for durations or timeouts; use the monotonic clock.

Ordering events across machines by wall-clock timestamp silently loses writes under last-write-wins.

🧪 Practice

Compute the divergence between two clocks drifting +40 ppm and -40 ppm after one week without synchronization.

Find every use of wall-clock time in a codebase you know and classify each as correct or as a latent bug.

Interview: A distributed rate limiter occasionally allows double the configured rate. What do you suspect? (Hint: which clock did the window boundaries come from, and what happens when one of them steps?)

NTP and Clock Synchronization

Since clocks drift, they must be corrected, and the Network Time Protocol is how that is done almost everywhere. Understanding what NTP can and cannot guarantee determines how much you are allowed to trust a timestamp.

NTP works hierarchically. Stratum 0 is a reference source — an atomic clock or GPS receiver. Stratum 1 servers are directly attached to one. Each subsequent stratum synchronizes from the one above, with accuracy degrading at each hop. A client repeatedly exchanges timestamped messages with its servers and computes both the round-trip delay and the estimated offset.

 THE NTP EXCHANGE AND ITS FUNDAMENTAL LIMITATION

   client                              server
     |  t1 = send time                   |
     |---------------------------------> | t2 = receive time
     |                                   | t3 = reply time
     | <---------------------------------|
     |  t4 = receive time                |

   delay  = (t4 - t1) - (t3 - t2)
   offset = ((t2 - t1) + (t3 - t4)) / 2

   The offset formula ASSUMES the network path is symmetric -- that the
   request took as long as the reply. When it is not (a congested uplink, an
   asymmetric route), the error is half the asymmetry, and NTP cannot detect
   it. This is why NTP gives you accuracy of milliseconds, not microseconds.

Two correction modes exist, and the difference matters for application correctness:

Mode	Behaviour	Risk
Slew	Speeds up or slows the clock to converge gradually	Slow to correct large offsets
Step	Jumps the clock to the correct value	Time moves BACKWARDS — breaks durations, can fire timers twice or never

# What a healthy client looks like, and what to check when it is not.
chronyc tracking
# Reference ID    : 10.0.0.1 (ntp.internal)
# Stratum         : 3
# System time     : 0.000234 seconds fast of NTP time
# RMS offset      : 0.000512 seconds        <- your real accuracy: ~0.5 ms
# Root dispersion : 0.001204 seconds        <- accumulated uncertainty
# Leap status     : Normal

chronyc sources -v      # confirm several sources agree; one bad source that
                        # all your servers share is a correlated failure

Two well-known hazards deserve specific mention.

Leap seconds. Occasionally a second is inserted to keep atomic time aligned with Earth's rotation, producing a 23:59:60 that much software cannot represent. Historic leap seconds have caused widespread outages. The modern fix is leap smearing — spreading the extra second across many hours so no discontinuity occurs — which every major cloud provider now does. The catch is that a smeared server and an unsmeared one disagree by up to a second during the smear window, so a fleet must use one policy consistently.

Virtualization. A VM's clock stops advancing correctly while the VM is paused, migrated, or descheduled under host contention. Guests must use host-provided time sources, and even then a resumed VM can need a step correction — reintroducing exactly the backwards jump that breaks durations.

The engineering conclusion is that NTP gives you bounded uncertainty, not correct time, and the bound is milliseconds at best. That is why Google built TrueTime for Spanner: rather than pretending the clock is exact, TrueTime returns an interval [earliest, latest] and the system waits out the uncertainty before committing, guaranteeing that a transaction's timestamp really is in the past everywhere.

# The idea behind commit-wait, which is what TrueTime buys.
def commit_with_uncertainty(txn, uncertainty_ms=7):
    commit_ts = now_latest()             # the LATEST time it might be
    apply_writes(txn, timestamp=commit_ts)
    # Wait until commit_ts is definitely in the past on EVERY node, so no
    # other node can later assign an earlier timestamp to a later event.
    time.sleep(uncertainty_ms / 1000.0)
    make_visible(txn)
# The cost of external consistency is exactly this wait -- a few milliseconds
# per transaction, paid deliberately, in exchange for globally correct
# ordering. Spanner's investment in GPS and atomic clocks exists to make the
# uncertainty small enough that this wait is affordable.

Key Takeaways

NTP corrects drift hierarchically but assumes symmetric network paths, which bounds accuracy to milliseconds.

Step corrections move the clock backwards and break anything measuring durations; prefer slewing and always use monotonic clocks for intervals.

Leap seconds and VM pause/migration are the two classic sources of sudden clock discontinuity; smear policy must be fleet-wide.

NTP provides bounded uncertainty, not exact time — systems needing correct ordering must make the uncertainty explicit and wait it out.

🧪 Practice

Compute the NTP offset and delay for t1=100.000, t2=100.030, t3=100.032, t4=100.070, and state the assumption you relied on.

Explain what happens to a 30-second lease when the holder's clock steps forward by 60 seconds.

Interview: Why does Spanner need GPS and atomic clocks when NTP exists? (Hint: the protocol does not remove uncertainty — what does Spanner do with the uncertainty, and what does its size cost?)

Lamport Timestamps

Since physical clocks cannot order events across machines, Lamport proposed abandoning physical time entirely. A logical clock is a counter that captures causality rather than duration — it does not tell you when something happened, only what it could have been influenced by.

The algorithm, from Lamport's 1978 paper, is three rules and a single integer per process:

Before any local event, increment your counter.
When sending a message, increment and attach your counter.
On receiving a message, set your counter to max(local, received) + 1.

That max is the entire mechanism. It propagates knowledge forward: a process that receives a message can never again issue a timestamp lower than the sender's at the moment of sending, so causality is preserved in the numbers.

class LamportClock:
    def __init__(self):
        self.time = 0

    def local_event(self):
        self.time += 1              # rule 1
        return self.time

    def send(self):
        self.time += 1              # rule 2
        return self.time            # attach this to the message

    def receive(self, sender_time):
        # rule 3: adopt the sender's knowledge, then advance past it. This is
        # what guarantees the receiver's next timestamp exceeds the sender's.
        self.time = max(self.time, sender_time) + 1
        return self.time

 LAMPORT CLOCKS IN ACTION

  P1: 1 -- 2 ---------------- 3
           |  msg(2)          |
           v                  |
  P2:      3 -- 4 ----------- |
                | msg(4)      |
                v             v
  P3: 1 ------- 5 ----------- 6

  P2's receive: max(0, 2) + 1 = 3.  Its next local event is 4.
  P3's receive: max(1, 4) + 1 = 5.

  Guarantee: if A happens-before B, then L(A) < L(B).
  NOT guaranteed: if L(A) < L(B), then A happens-before B.

That asymmetry is the crucial limitation. Lamport timestamps give a one-way implication. A lower timestamp does not mean causality — two concurrent events on different processes can have any relative timestamps, and the numbers cannot tell you they were concurrent.

# The limitation, demonstrated.
# P1 does three local events:  L = 1, 2, 3
# P2 does one local event:     L = 1
# These processes never communicated, so ALL FOUR events are concurrent.
#
# Yet P1's event 1 and P2's event 1 have the SAME timestamp, and P1's event 3
# has a HIGHER timestamp than P2's event 1 despite no causal relationship.
#
# Lamport timestamps can prove causality is POSSIBLE. They can never prove
# two events were concurrent. Vector clocks (next topic) close exactly this
# gap, at the cost of O(N) metadata instead of O(1).

Their practical value comes from the total order they produce when ties are broken deterministically. Sorting by (lamport_time, node_id) gives every node the same ordering of all events, and that order is consistent with causality — which is exactly what a replicated state machine needs.

def total_order_key(event):
    # Tie-break by node ID to make the order total and identical everywhere.
    # Any deterministic tiebreak works; every node must use the SAME one.
    return (event.lamport_time, event.node_id)

ordered = sorted(all_events, key=total_order_key)
# Every replica applying `ordered` reaches the same final state, and no
# effect is ever applied before its cause. The order is somewhat arbitrary
# for concurrent events -- which is fine, because by definition nothing
# observed them in a particular order.

Lamport clocks appear in production far more often than their age suggests: as sequence numbers in replicated logs, as the ordering mechanism inside consensus protocols, and as the basis of hybrid logical clocks. They cost one integer, which is why they remain the default when full causality tracking is too expensive.

Key Takeaways

A Lamport clock is a per-process counter advanced by max(local, received) + 1 on receipt, capturing causality without physical time.

It guarantees that if A happens-before B then L(A) < L(B), but the converse does not hold.

It cannot detect concurrency — two unrelated events may have any relative timestamps.

Breaking ties by node ID yields a total order consistent with causality, which is what replicated state machines need.

🧪 Practice

Trace Lamport timestamps for three processes exchanging five messages of your choosing, and identify a pair of concurrent events with ordered timestamps.

Explain why max(local, received) + 1 is required rather than received + 1.

Interview: Two events have Lamport timestamps 7 and 9. What can you conclude? (Hint: state the implication in the direction it actually runs.)

Vector Clocks

Vector clocks fix the one thing Lamport timestamps cannot do: detect concurrency. Instead of one counter, each process keeps a vector with one entry per process — its own count plus its knowledge of everyone else's.

The idea is that a vector is a summary of everything a process has seen. If process A's vector dominates process B's in every position, A knows everything B knew, so B's event happened before A's. If each vector is ahead in some position and behind in another, neither knows about the other's event, so they are concurrent by definition.

class VectorClock:
    def __init__(self, node_id, nodes):
        self.node_id = node_id
        self.vec = {n: 0 for n in nodes}

    def local_event(self):
        self.vec[self.node_id] += 1        # only ever increment YOUR OWN entry
        return dict(self.vec)

    def send(self):
        self.vec[self.node_id] += 1
        return dict(self.vec)              # ship the whole vector

    def receive(self, other):
        # Take the elementwise maximum: adopt everything the sender knew.
        for n, c in other.items():
            self.vec[n] = max(self.vec.get(n, 0), c)
        self.vec[self.node_id] += 1        # then account for this receive event
        return dict(self.vec)

def compare(a, b):
    """Returns 'before', 'after', 'concurrent', or 'equal'."""
    a_le_b = all(a.get(n, 0) <= b.get(n, 0) for n in set(a) | set(b))
    b_le_a = all(b.get(n, 0) <= a.get(n, 0) for n in set(a) | set(b))
    if a_le_b and b_le_a: return "equal"
    if a_le_b:            return "before"        # a happens-before b
    if b_le_a:            return "after"
    return "concurrent"   # neither dominates -> genuinely parallel, and this
                          # is the answer Lamport clocks cannot give

 DETECTING A CONFLICT THAT LAMPORT CLOCKS WOULD MISS

  A writes:  {A:1, B:0, C:0}
  B writes:  {A:0, B:1, C:0}

  compare -> A is ahead in position A, B is ahead in position B
          -> CONCURRENT -> a genuine write-write conflict

  Now B receives A's update and writes again: {A:1, B:2, C:0}
  compare with A's {A:1, B:0, C:0}
          -> B dominates in every position -> AFTER
          -> not a conflict; B's write supersedes A's, safely

This is what lets a Dynamo-style store distinguish "this write supersedes that one" from "these two writes genuinely conflict and someone must decide". A conflict detected is a conflict that can be resolved intentionally — by application logic, by presenting both siblings to the client, or by a CRDT merge — rather than silently discarded by last-write-wins.

# Conflict handling in a versioned store.
def put(key, value, context_clock):
    existing = store.get(key)
    if existing is None:
        return store.write(key, [(value, clock.local_event())])

    siblings = []
    for old_value, old_clock in existing:
        rel = compare(context_clock, old_clock)
        if rel == "after":
            continue                        # our write supersedes this one
        if rel == "concurrent":
            siblings.append((old_value, old_clock))   # keep BOTH; resolve later
    siblings.append((value, clock.local_event()))
    return store.write(key, siblings)
# The shopping-cart classic: two concurrent adds produce two siblings, and
# the merge is a set union -- so nothing the customer added is ever lost.

The cost is metadata that grows with the number of writers: O(N) per value for N nodes. In a cluster with hundreds of nodes and clients writing directly, the clock can dwarf the data. Real systems bound it in three ways:

Technique	Approach
Per-key vectors	Track only the nodes that wrote this key
Client-side IDs	One entry per client session, not per storage node
Dotted version vectors	Precise sibling tracking with bounded size
Pruning	Drop the oldest entries past a threshold, accepting rare false conflicts

Property	Lamport	Vector
Size	O(1)	O(N)
Detects causality	One-way only	Yes
Detects concurrency	No	Yes
Gives a total order	Yes (with tiebreak)	Partial order only

Key Takeaways

A vector clock holds one counter per process, so comparing vectors reveals whether events are ordered or concurrent.

Domination in every position means happens-before; mutual non-domination means genuine concurrency.

Detecting concurrency is what allows a conflict to be resolved deliberately rather than silently lost.

The O(N) metadata cost is real; bound it with per-key vectors, client-scoped entries, or dotted version vectors.

🧪 Practice

Given {A:2,B:1,C:0} and {A:1,B:3,C:0}, determine the relationship and explain your reasoning position by position.

Implement the shopping-cart sibling merge and show that no item is lost under concurrent adds from two devices.

Interview: Why did Riak expose siblings to the client instead of resolving conflicts itself? (Hint: what does the database know about which of two concurrent values is "right"?)

Hybrid Logical Clocks

Logical clocks capture causality but produce numbers with no relation to real time — you cannot ask "what did this key look like at 14:00?" of a Lamport timestamp. Physical clocks relate to real time but cannot be trusted for ordering. Hybrid logical clocks (HLC) combine both: a timestamp that is close to physical time and guaranteed consistent with causality.

An HLC timestamp is a pair (l, c): l is a physical-time component that only ever moves forward, and c is a logical counter that breaks ties when multiple events share the same l. The update rules keep l close to wall-clock time while never allowing it to violate happens-before.

class HybridLogicalClock:
    def __init__(self):
        self.l = 0      # physical component (ms), monotonically non-decreasing
        self.c = 0      # logical counter, breaks ties within the same ms

    def now(self):
        return int(time.time() * 1000)

    def send_or_local(self):
        pt = self.now()
        if pt > self.l:
            self.l, self.c = pt, 0        # physical time advanced: use it,
        else:                             # reset the counter
            self.c += 1                   # clock has not ticked (or went
                                          # backwards): advance logically
        return (self.l, self.c)

    def receive(self, m_l, m_c):
        pt = self.now()
        # Take the maximum of: our clock, the message's clock, physical time.
        new_l = max(self.l, m_l, pt)
        if new_l == self.l == m_l:
            self.c = max(self.c, m_c) + 1     # same ms on both sides
        elif new_l == self.l:
            self.c += 1                        # ours was ahead
        elif new_l == m_l:
            self.c = m_c + 1                   # theirs was ahead
        else:
            self.c = 0                         # physical time won: fresh start
        self.l = new_l
        return (self.l, self.c)

 WHY HLC STAYS CLOSE TO PHYSICAL TIME

  node A (clock 100 ms FAST) sends at (1000500, 0)
  node B (clock correct, real time 1000200) receives:
      new_l = max(1000200, 1000500, 1000200) = 1000500
      -> B's HLC jumps to A's value to preserve causality

  B's HLC is now 300 ms ahead of its physical clock -- but BOUNDED by the
  maximum clock skew in the cluster, because that is the largest jump any
  message can induce. Unlike a Lamport clock, it cannot run away: once
  physical time catches up, the l component tracks it again.

The properties that make HLCs useful in production databases:

Property	Lamport	Physical	HLC
Consistent with causality	Yes	No	Yes
Close to wall-clock time	No	Yes	Yes
Constant size	Yes	Yes	Yes
Bounded divergence from real time	No	—	Yes (by max skew)
Detects concurrency	No	No	No

# The capability HLCs enable: meaningful time-travel queries on a causally
# consistent timestamp.
def read_as_of(key, wall_clock_ms):
    # Because the l component tracks physical time within the skew bound, a
    # human-meaningful time maps onto a causally sound snapshot.
    return store.read_version_at(key, hlc=(wall_clock_ms, 0))

# SELECT * FROM orders AS OF SYSTEM TIME '2026-08-25 14:00:00';
#
# This is exactly what CockroachDB and YugabyteDB do. It is impossible with
# pure Lamport timestamps (the numbers mean nothing to a human) and unsafe
# with pure physical timestamps (the snapshot could violate causality).

HLCs are used by CockroachDB, YugabyteDB, MongoDB (for its cluster time), and several others. They occupy a genuine sweet spot: the causal safety of logical clocks, the interpretability of physical clocks, constant size, and — unlike Spanner's TrueTime — no requirement for specialized GPS or atomic-clock hardware. The trade is that HLC does not shrink clock uncertainty, so systems using it generally offer serializable snapshot isolation rather than Spanner's external consistency.

Key Takeaways

An HLC pairs a monotonic physical component with a logical counter, giving timestamps that are both causally correct and human-meaningful.

Its divergence from real time is bounded by the cluster's maximum clock skew, so it cannot run away like a Lamport clock.

It has constant size and needs no special hardware, which is why modern distributed SQL databases adopt it.

Like Lamport clocks, it produces a total order and cannot detect concurrency.

🧪 Practice

Trace HLC values for two nodes with 50 ms of skew exchanging three messages.

Explain why l can exceed local physical time and what bounds the excess.

Interview: Why can CockroachDB support AS OF SYSTEM TIME queries when a Lamport-clock system cannot? (Hint: what would you pass as the argument?)

Happens-Before Relation

The happens-before relation, written A → B, is the formal definition of potential causality that underlies everything in this subchapter. Lamport defined it in 1978, and every clock mechanism above exists to approximate or capture it.

It is defined by three rules:

If A and B are events in the same process and A comes first, then A → B.
If A is the sending of a message and B is its receipt, then A → B.
Transitivity: if A → B and B → C, then A → C.

If neither A → B nor B → A, the events are concurrent, written A || B.

The crucial subtlety is in the word potential. Happens-before does not mean A caused B; it means A could have influenced B, because information could have flowed from one to the other. Two events may be causally related in the real world — I phone you and you act on it — while being concurrent as far as the system is concerned, because the influence travelled outside the system. That gap is why "external consistency" is a stronger and separate requirement.

 THE HAPPENS-BEFORE LATTICE

  P1: a ------> b ---------> c
                 \  msg
                  \
  P2:      d ----> e ------> f
                    \  msg
                     \
  P3: g -------------> h --> i

  a -> b -> c          (rule 1, within P1)
  b -> e               (rule 2, message)
  b -> e -> f          (rule 3, transitivity)
  b -> e -> h -> i     (rule 3, a longer chain)

  a || d               (no path between them: CONCURRENT)
  c || f               (c happens after b, but nothing links c to f)
  a || g               (different processes, no communication)

  Concurrency is the ABSENCE of a path, not a statement about wall-clock
  simultaneity. c and f may occur hours apart and still be concurrent.

# Building the relation explicitly, which is what a debugger or a tracing
# system does when it reconstructs a distributed execution.
def happens_before(a, b, events):
    """Is there a causal path from a to b?"""
    if a.process == b.process:
        return a.seq < b.seq                       # rule 1
    # Rules 2 and 3: walk the graph of message edges and process order.
    return _reachable(a, b, build_edges(events))

def build_edges(events):
    edges = defaultdict(set)
    by_process = defaultdict(list)
    for e in events:
        by_process[e.process].append(e)
    for proc_events in by_process.values():        # rule 1 edges
        for earlier, later in zip(proc_events, proc_events[1:]):
            edges[earlier.id].add(later.id)
    for e in events:                               # rule 2 edges
        if e.type == "receive":
            edges[e.matching_send_id].add(e.id)
    return edges                                   # rule 3 is reachability

def is_concurrent(a, b, events):
    return not happens_before(a, b, events) and not happens_before(b, a, events)

Happens-before is a partial order, not a total one — and that is not a deficiency, it is a truthful description of reality. Concurrent events genuinely have no order; imposing one requires either coordination (making them not concurrent) or an arbitrary tiebreak (declaring an order that means nothing).

The relation to the clock mechanisms is exact and worth stating as a summary, because it explains what each one is for:

Mechanism	Relationship to happens-before
Lamport clock	`A → B` implies `L(A) < L(B)`; the converse does not hold
Vector clock	`A → B` if and only if `V(A) < V(B)`; captures it exactly
HLC	Same guarantee as Lamport, plus a physical-time approximation
Physical clock	No guarantee at all — skew can invert the order

Practically, happens-before is what you reason with when deciding whether two operations need coordination. If A → B in your design, ordering is already guaranteed by the message flow and no synchronization is needed. If A || B, either the operations must commute (so order does not matter), or you must introduce coordination to create an ordering, or you must detect the conflict and resolve it. There is no fourth option, and recognizing which case you are in is most of distributed design.

Key Takeaways

A → B means information could have flowed from A to B, via process order, message passing, or transitivity.

Concurrent means no causal path exists — not that the events were simultaneous in wall-clock terms.

It is a partial order because concurrent events genuinely have no order; forcing one requires coordination or an arbitrary tiebreak.

Vector clocks capture happens-before exactly; Lamport clocks and HLCs capture only the forward implication.

🧪 Practice

For the lattice diagram above, list every pair involving d and classify each as ordered or concurrent.

Explain why happens-before captures potential rather than actual causality, and give an example where the two differ.

Interview: Two writes to the same key are concurrent. What are your options? (Hint: there are exactly three, and one of them changes the operations rather than the system.)

8.4 Consensus and Coordination

Consensus is the problem of getting a group of unreliable machines to agree on a single value. Nearly every hard coordination task — electing a leader, holding a lock, committing a transaction, changing cluster membership — reduces to it.

Leader Election

Many distributed problems become dramatically simpler if exactly one node is designated to make decisions. A single leader can serialize writes, assign sequence numbers, hold the authoritative view of state, and coordinate others — turning a distributed agreement problem into a local one.

The difficulty is not electing a leader; it is guaranteeing there is at most one at a time when nodes cannot reliably tell whether a peer is dead or merely slow. FLP tells us that determination cannot be made perfectly, so every leader election protocol is built to remain correct when the determination is wrong.

 WHY "AT MOST ONE" IS THE HARD PART

  Leader L is GC-pausing for 8 seconds.
  Followers stop receiving heartbeats and elect L2.
  L wakes up, unaware that any time passed, and continues acting as leader.

  time --->
   L : [leader]-----[  GC pause  ]----[still believes it is leader]----->
   F1: heartbeats.....(silence)......[elect L2]......................
   L2:                                [leader]------------------------>

                                       ^^^^^^ TWO LEADERS ^^^^^^

  Both may write. Both may respond to clients. The system has split brain
  without any network partition at all -- just a slow process.

Three mechanisms combine to make this safe.

Majority quorum. A candidate needs votes from more than half the cluster. Since two majorities of the same set must intersect, and each node votes at most once per term, at most one leader can be elected per term.

Terms (epochs). Every election increments a monotonically increasing number. Messages carry the term of their sender, and any node seeing a higher term immediately steps down. This is how a returning old leader learns it was replaced — the first response it gets carries a term it does not recognize.

Fencing tokens. External systems reject writes carrying a term lower than the highest they have seen, so a stale leader is stopped by the storage layer even before it discovers its own obsolescence.

# Term-based election, the core of Raft's approach.
class Node:
    def __init__(self, node_id, peers):
        self.id, self.peers = node_id, peers
        self.term = 0
        self.voted_for = None            # at most ONE vote per term
        self.state = "follower"
        self.election_timeout = self._random_timeout()

    def _random_timeout(self):
        # Randomized so nodes do not all time out together and split the vote.
        # Without this, repeated split elections can stall the cluster.
        return random.uniform(0.150, 0.300)

    def start_election(self):
        self.term += 1                   # new term
        self.state = "candidate"
        self.voted_for = self.id         # vote for yourself
        votes = 1
        for peer in self.peers:
            granted, peer_term = peer.request_vote(self.term, self.id,
                                                   self.last_log_index,
                                                   self.last_log_term)
            if peer_term > self.term:
                self.step_down(peer_term)     # someone is more current
                return
            votes += granted
        if votes > (len(self.peers) + 1) // 2:
            self.become_leader()

    def request_vote(self, term, candidate_id, last_log_index, last_log_term):
        if term < self.term:
            return False, self.term           # stale candidate
        if term > self.term:
            self.step_down(term)              # adopt the newer term
        # Grant only if we have not voted this term AND the candidate's log
        # is at least as up to date as ours -- this second condition is what
        # prevents a node missing committed entries from becoming leader.
        up_to_date = (last_log_term, last_log_index) >= (self.last_log_term,
                                                         self.last_log_index)
        if self.voted_for in (None, candidate_id) and up_to_date:
            self.voted_for = candidate_id
            return True, self.term
        return False, self.term

The parameter that governs behaviour is the election timeout, and it is a direct trade:

def election_tradeoff(heartbeat_ms, timeout_ms, p99_gc_pause_ms):
    return {
        "detection_time_ms": timeout_ms,
        "missed_heartbeats_tolerated": timeout_ms // heartbeat_ms,
        # If the timeout is shorter than a plausible GC pause, healthy leaders
        # get deposed regularly -- each election costing a window of
        # unavailability for no reason.
        "spurious_elections_likely": timeout_ms < p99_gc_pause_ms * 1.5,
    }

print(election_tradeoff(heartbeat_ms=50, timeout_ms=200, p99_gc_pause_ms=400))
# {'detection_time_ms': 200, 'missed_heartbeats_tolerated': 4,
#  'spurious_elections_likely': True}
#
# A 200 ms timeout on a JVM with 400 ms GC pauses guarantees flapping
# leadership. Either tune the GC or lengthen the timeout -- but know that
# lengthening it directly lengthens failover time.

Approach	Guarantees	Notes
Consensus-based (Raft/Paxos)	Safe: one leader per term	Needs a majority to be alive
Lease from a coordinator	Safe if clock drift is bounded	Depends on ZooKeeper/etcd
Bully algorithm	Simple; assumes reliable detection	Not partition-safe
Ring-based	Low message count	Fragile to node failure

Key Takeaways

A single leader simplifies coordination; the hard guarantee is that at most one exists at any time.

Majority quorum plus one vote per term makes two leaders in the same term impossible.

Monotonic terms let a returning stale leader detect it was replaced, and fencing tokens stop it before it does.

Election timeout trades failover speed against spurious elections — it must exceed plausible GC pauses.

🧪 Practice

Explain why randomized election timeouts are necessary and what happens without them.

Show how the "candidate's log must be up to date" rule prevents committed data loss.

Interview: A leader GC-pauses for 10 seconds and returns. What stops it from corrupting state? (Hint: two independent mechanisms — one it learns from its peers, one enforced without its cooperation.)

Paxos Overview

Paxos, published by Lamport in 1998, was the first proven-correct solution to consensus under asynchrony with crash failures. It is famous for being difficult to understand and is the theoretical ancestor of most systems in production.

The problem it solves precisely: a set of nodes must agree on one value, such that (safety) only a proposed value is chosen and all nodes learn the same one, and (liveness) some value is eventually chosen if enough nodes are alive. Given FLP, Paxos guarantees safety unconditionally and liveness only when the network behaves.

There are three roles, often played by the same physical nodes: proposers suggest values, acceptors vote, and learners observe the outcome.

Basic Paxos runs in two phases:

 PAXOS: TWO PHASES

 PHASE 1 (Prepare / Promise)
   proposer -> acceptors:  PREPARE(n)              n = a unique proposal number
   acceptor:  if n > any n seen before:
                promise never to accept a proposal numbered < n
                reply PROMISE(n, highest accepted value so far, if any)

 PHASE 2 (Propose / Accept)
   proposer:  if a majority promised:
                if ANY promise carried an accepted value ->
                    MUST propose the value with the highest proposal number
                else -> may propose its own value
   proposer -> acceptors:  ACCEPT(n, value)
   acceptor:  accept unless it has promised a higher n
   -> once a majority accepts, the value is CHOSEN, permanently

The subtle heart of the algorithm is the rule in bold: a proposer that learns of any previously accepted value must adopt it rather than proposing its own. That is what makes the choice permanent — once a value could have been chosen, every future proposal carries it forward, so no later round can choose differently.

class Acceptor:
    def __init__(self):
        self.promised_n = None       # highest proposal number promised
        self.accepted_n = None       # proposal number of the accepted value
        self.accepted_value = None

    def prepare(self, n):
        if self.promised_n is not None and n <= self.promised_n:
            return None              # reject: we promised a higher number
        self.promised_n = n
        # Report what we already accepted, so the proposer can preserve it.
        return {"promised": n, "accepted_n": self.accepted_n,
                "accepted_value": self.accepted_value}

    def accept(self, n, value):
        if self.promised_n is not None and n < self.promised_n:
            return False             # a newer proposal has superseded this one
        self.promised_n = n
        self.accepted_n, self.accepted_value = n, value
        return True

class Proposer:
    def propose(self, acceptors, my_value):
        n = self.next_proposal_number()          # globally unique, increasing
        promises = [p for p in (a.prepare(n) for a in acceptors) if p]
        if len(promises) <= len(acceptors) // 2:
            return None                          # no majority; retry with a
                                                 # higher n
        # THE CRITICAL RULE: if any acceptor already accepted something, that
        # value may already have been chosen. Adopt the one with the highest
        # accepted_n. Only if none exists may we propose our own.
        prior = [p for p in promises if p["accepted_value"] is not None]
        value = (max(prior, key=lambda p: p["accepted_n"])["accepted_value"]
                 if prior else my_value)
        accepts = sum(a.accept(n, value) for a in acceptors)
        return value if accepts > len(acceptors) // 2 else None

Basic Paxos agrees on a single value. Real systems need an ordered sequence of values — a replicated log — which is Multi-Paxos: run an instance per log slot, and observe that a stable proposer can skip Phase 1 for subsequent slots. That optimization is exactly a leader, and it reduces the steady state to one round trip per entry.

Paxos's two well-known problems in practice:

Duelling proposers. Two proposers repeatedly outbid each other in Phase 1, and neither reaches Phase 2. Safety holds; liveness does not. This is FLP made concrete, and the fix is to elect a distinguished proposer — again, a leader.

Underspecification. The original paper describes single-decree consensus and leaves membership changes, log compaction, and recovery to the implementer. Two teams implementing "Paxos" can produce systems that differ substantially, which is the practical complaint that motivated Raft.

Variant	Contribution
Basic Paxos	Agreement on one value
Multi-Paxos	A log of values; a stable leader skips Phase 1
Fast Paxos	One fewer round trip in the common case, larger quorums
Cheap Paxos	Uses auxiliary nodes only during failures
EPaxos	Leaderless; commutative commands commit in one round trip

Key Takeaways

Paxos guarantees safety unconditionally and liveness only when the network cooperates, exactly as FLP requires.

Phase 1 collects promises and any previously accepted value; Phase 2 proposes and gets it accepted by a majority.

The rule that a proposer must adopt any previously accepted value is what makes a chosen value permanent.

Multi-Paxos with a stable leader is the practical form, reducing steady state to one round trip per log entry.

🧪 Practice

Trace Basic Paxos with three acceptors and two proposers whose messages interleave, and show that only one value is chosen.

Explain what breaks if a proposer ignores previously accepted values and always proposes its own.

Interview: Why does Multi-Paxos skip Phase 1 for most entries? (Hint: what is Phase 1 actually establishing, and does it need re-establishing each time?)

Raft Consensus

Raft was designed in 2014 with an explicit goal that no prior protocol had: understandability. It provides the same guarantees as Multi-Paxos, but decomposes the problem into three independent pieces — leader election, log replication, and safety — so that each can be reasoned about separately.

Raft's central simplification is a strong leader. All client requests go to the leader, log entries flow only from leader to followers, and a follower never overwrites the leader's decisions. Where Paxos allows any proposer to act at any time, Raft insists on one direction of flow, and much of its clarity comes from that restriction.

Every node is in one of three states:

 RAFT STATE MACHINE

              times out,                 receives votes
              starts election            from a majority
   +----------+ ------> +-----------+ ------> +--------+
   | FOLLOWER |         | CANDIDATE |         | LEADER |
   +----------+ <------ +-----------+ <------ +--------+
        ^   discovers a current leader   discovers a
        |   or a higher term             higher term
        |                                       |
        +---------------------------------------+

  Terms act as a logical clock. Every message carries a term; any node
  seeing a higher term immediately reverts to follower.

Log replication is the steady-state path:

 LOG REPLICATION AND COMMITMENT

  client -> leader:  SET x=5
  leader appends to its own log (uncommitted), then:

  leader   [1:a][2:b][3:x=5]
  follower [1:a][2:b][3:x=5]   <- acknowledged
  follower [1:a][2:b][3:x=5]   <- acknowledged
  follower [1:a][2:b]          <- lagging, will be caught up
  follower [1:a][2:b]          <- lagging

  A majority (3 of 5) has entry 3 -> the leader marks it COMMITTED,
  applies it to its state machine, and replies to the client. Followers
  learn of the commit on the next AppendEntries and apply it themselves.

def append_entries(self, term, leader_id, prev_log_index, prev_log_term,
                   entries, leader_commit):
    """Follower side: also serves as the heartbeat when entries is empty."""
    if term < self.term:
        return False, self.term                  # stale leader

    self.reset_election_timer()                  # a valid leader exists
    if term > self.term:
        self.step_down(term)

    # CONSISTENCY CHECK: refuse unless our log matches at prev_log_index.
    # This single check inductively guarantees that if two logs agree at an
    # index and term, they agree on EVERYTHING before it -- the Log Matching
    # Property, which is why followers can never silently diverge.
    if prev_log_index >= len(self.log) or \
       (prev_log_index >= 0 and self.log[prev_log_index].term != prev_log_term):
        return False, self.term                  # leader will retry with an
                                                 # earlier index until we match

    # Delete any conflicting suffix, then append. The leader's log is truth.
    self.log = self.log[:prev_log_index + 1] + entries

    if leader_commit > self.commit_index:
        self.commit_index = min(leader_commit, len(self.log) - 1)
        self.apply_committed()                   # apply to the state machine
    return True, self.term

Raft's safety rests on two rules that are easy to state and carry the whole argument:

Election restriction. A node only grants a vote if the candidate's log is at least as up to date as its own. Since a committed entry is on a majority, and any winning candidate needs a majority, the two majorities intersect — so any elected leader necessarily has every committed entry.

Commit only from the current term. A leader never marks an entry from a previous term as committed by counting replicas alone; it commits an entry of its own term, which carries the earlier ones with it. Without this rule there is a subtle scenario where a replicated-but-uncommitted entry can be overwritten after appearing safe.

def advance_commit_index(self):
    """Leader: find the highest index replicated on a majority."""
    for n in range(len(self.log) - 1, self.commit_index, -1):
        replicated = 1 + sum(1 for p in self.peers if self.match_index[p] >= n)
        # The second condition is the subtle safety rule. Entries from older
        # terms are committed only INDIRECTLY, by committing a current-term
        # entry that follows them.
        if replicated > (len(self.peers) + 1) // 2 and self.log[n].term == self.term:
            self.commit_index = n
            self.apply_committed()
            return

Aspect	Paxos	Raft
Leader	Optional optimization	Mandatory and central
Log flow	Any direction	Leader to followers only
Log holes	Permitted	Forbidden — logs are contiguous
Membership change	Unspecified	Joint consensus, specified
Specification	Single-decree; gaps	Complete, implementable

Raft powers etcd (and therefore Kubernetes), Consul, CockroachDB, TiKV, and many others. It is the default choice for new systems needing consensus, not because it is faster — the performance is comparable — but because a protocol engineers can reason about produces implementations with fewer bugs.

Key Takeaways

Raft provides Multi-Paxos guarantees while decomposing the problem into election, replication, and safety.

A strong leader and a leader-to-follower-only log flow are the source of its comparative simplicity.

The log matching consistency check makes divergence impossible; the election restriction guarantees a new leader holds every committed entry.

A leader must commit an entry from its own term before earlier entries count as committed.

🧪 Practice

Trace a five-node cluster where the leader fails after replicating an entry to two followers. Which nodes can win the next election?

Explain the Log Matching Property and why the prev_log_index check is sufficient to maintain it.

Interview: Why can a Raft leader not commit a previous term's entry by counting replicas? (Hint: construct the sequence where that entry is later overwritten by a new leader.)

Zab and ZooKeeper

ZooKeeper is a coordination service — a small, strongly consistent, highly available store designed to hold the metadata that other distributed systems coordinate through. Its consensus protocol is Zab (ZooKeeper Atomic Broadcast), designed specifically for primary-backup replication of a state machine.

The reason a separate coordination service exists is economic. Consensus is hard to implement correctly and expensive to run, so rather than embedding it in every application, you run one well-tested cluster and let everything else delegate its hard coordination problems to it. Kafka (historically), HBase, Solr, and many others took exactly this route.

Zab's distinctive guarantee versus Paxos is primary order: writes from a single primary are delivered in the order that primary issued them, and a new primary's writes come after all of the previous one's. This is stronger than Paxos's guarantee and simpler for building a primary-backup replicated state machine.

 ZAB PHASES

  1. DISCOVERY   followers report their latest epoch and history; the
                 prospective leader learns the most up-to-date state
  2. SYNCHRONIZATION  the leader brings every follower to an identical
                 history -- this is what enforces primary order
  3. BROADCAST   normal operation: two-phase proposal per write

  BROADCAST detail (per write):
     leader -> followers: PROPOSE(zxid, txn)
     followers          : write to disk, reply ACK
     leader             : on a QUORUM of ACKs -> COMMIT
     leader -> followers: COMMIT(zxid)

  zxid = (epoch << 32) | counter -- a 64-bit id where the high bits are the
  leadership epoch, so ordering across leadership changes is automatic.

The data model is a hierarchical namespace of znodes, each holding a small amount of data (typically under 1 MB) plus metadata. Three znode flavours provide the coordination primitives:

Type	Lifetime	Used for
Persistent	Until explicitly deleted	Configuration, cluster metadata
Ephemeral	Deleted when the session ends	Membership, liveness, locks
Sequential	Name gets a monotonic suffix	Queues, fair lock ordering

Ephemeral nodes are the key idea. Because the node vanishes automatically when the creating client's session expires, liveness detection requires no separate heartbeat protocol — the existence of the znode is the liveness signal.

from kazoo.client import KazooClient

zk = KazooClient(hosts="zk1:2181,zk2:2181,zk3:2181")
zk.start()

# Leader election via ephemeral sequential nodes: each contender creates one,
# and the lowest sequence number wins.
path = zk.create("/election/node-", ephemeral=True, sequence=True)
# -> "/election/node-0000000042"

def try_become_leader():
    children = sorted(zk.get_children("/election"))
    my_name = path.rsplit("/", 1)[1]
    if children[0] == my_name:
        return become_leader()
    # NOT the leader. Watch ONLY the node immediately ahead of us, never the
    # whole list -- watching all of them means every node wakes on every
    # change, which is the "herd effect" that makes naive implementations
    # collapse at scale.
    predecessor = children[children.index(my_name) - 1]
    zk.exists(f"/election/{predecessor}", watch=lambda e: try_become_leader())

Two ZooKeeper behaviours cause most production surprises.

Reads are not linearizable by default. Reads are served from the local server, which may lag. sync() before a read forces the server to catch up first. Writes are always linearizable; reads are sequentially consistent unless you ask for more.

Session expiry is decided by the cluster, not the client. A client that is GC-paused or partitioned may still believe its session is alive after the cluster has expired it and deleted its ephemeral nodes. Any lock built on ephemeral nodes therefore needs fencing, exactly as in the next topic.

# The correct read pattern when you need the latest value.
zk.sync("/config/database")          # force this server to catch up
data, stat = zk.get("/config/database")
# Without sync(), you may read a value from before a write that has already
# been acknowledged to another client.

ZooKeeper's role has narrowed as systems adopt embedded Raft — Kafka's KRaft mode removed its ZooKeeper dependency, and etcd occupies much of the same niche with a simpler API. It remains widely deployed, and its primitives (ephemeral nodes, watches, sequential ordering) are the vocabulary that later coordination systems inherited.

Key Takeaways

ZooKeeper is a coordination service that centralizes the hard consensus problem so applications do not each solve it.

Zab adds primary order to consensus, matching the needs of primary-backup replication.

Ephemeral znodes make liveness detection automatic; sequential nodes give fair ordering for locks and queues.

Reads are local and can be stale unless preceded by sync(), and session expiry is the cluster's decision, not the client's.

🧪 Practice

Implement a distributed queue with sequential znodes and describe how consumers avoid taking the same item.

Explain the herd effect in a naive lock implementation and how watching only the predecessor avoids it.

Interview: A ZooKeeper client is GC-paused for 40 seconds with a 30-second session timeout. What happens to its lock? (Hint: two parties disagree about whether it still holds it — who is right, and who acts on it?)

Distributed Locks and Leases

A distributed lock grants one client exclusive access to a resource across machines. It is one of the most requested and most misused primitives in distributed systems, because it looks like a mutex and behaves nothing like one.

A local mutex is held by a thread whose existence the operating system can verify absolutely. A distributed lock is held by a process on another machine that may have crashed, paused, or been partitioned away — and the lock service cannot tell which. Every distributed lock therefore has a timeout, and the timeout is exactly the flaw: if the holder is merely slow, the lock expires while it is still working, and two clients hold it at once.

 THE UNAVOIDABLE FAILURE OF A LOCK WITH A TIMEOUT

  client 1: acquire ---[ working ]---[ GC PAUSE 20s ]---[ write! ]
  lock:     [------- 30s lease -------]X expired
  client 2:                            acquire ---[ working ]---[ write! ]
                                                                   ^
                          BOTH clients write. Mutual exclusion is gone,
                          and neither client did anything wrong.

  Making the lease longer does not fix it -- it only makes recovery from a
  genuine crash slower. There is no timeout value that is correct.

The resolution is not a better lock. It is to stop relying on mutual exclusion for correctness and add fencing, so the protected resource itself rejects stale holders:

# Every lock acquisition returns a monotonically increasing fencing token.
def acquire_with_fence(resource, ttl_ms):
    token = coordinator.increment_and_get(f"fence:{resource}")   # monotonic
    if coordinator.try_lock(resource, holder=self.id, ttl_ms=ttl_ms):
        return token
    raise LockUnavailable()

def write_protected(resource, data, token):
    # The STORAGE layer enforces exclusion, not the lock service. A client
    # that lost its lease carries an old token and is rejected here, even
    # though it believes it still holds the lock.
    current = storage.get_max_token(resource)
    if token < current:
        raise FencedOut(f"token {token} superseded by {current}")
    storage.write(resource, data, token=token)

 FENCING MAKES THE LATE WRITE HARMLESS

  client 1: acquire -> token 33 ---[ pause ]--- write(token=33) -> REJECTED
  client 2:            acquire -> token 34 ---- write(token=34) -> accepted

  Storage has seen 34, so 33 is refused. Client 1's belief that it holds
  the lock no longer matters -- correctness does not depend on it.

The distinction that should govern every use of a distributed lock:

Purpose	Consequence of double acquisition	Requirement
Efficiency	Duplicate work, wasted resources	A plain lock is fine
Correctness	Corruption, double charge, data loss	Fencing tokens are mandatory

If the lock is an optimization — avoid two nodes generating the same report — a simple Redis lock is entirely adequate; the worst case is wasted CPU. If the lock protects correctness, a lock alone is never sufficient, no matter which service provides it.

# Redlock -- Redis's multi-instance lock algorithm -- and its caveat.
def redlock_acquire(resource, ttl_ms, instances):
    start = time.monotonic()
    token = str(uuid.uuid4())
    acquired = sum(1 for r in instances
                   if r.set(resource, token, nx=True, px=ttl_ms))
    elapsed_ms = (time.monotonic() - start) * 1000
    validity = ttl_ms - elapsed_ms - CLOCK_DRIFT_ALLOWANCE
    if acquired > len(instances) // 2 and validity > 0:
        return token, validity
    for r in instances:                       # failed: release everywhere
        release(r, resource, token)
    return None, 0
# Redlock's validity argument depends on bounded clock drift across the Redis
# instances. Martin Kleppmann's well-known critique is that this makes it a
# timing-dependent protocol, unsuitable for correctness-critical locking
# without fencing -- and with fencing, you did not need Redlock's guarantees.

Two further practices matter. Renew rather than over-provision: hold a short lease and extend it while demonstrably alive, so a genuine crash releases quickly while a long job is not cut off. And release only your own lock: deleting a key without checking the token can release a lock another client now holds, which is a classic bug in hand-rolled Redis locks.

Key Takeaways

Distributed locks need timeouts, and any timeout allows a paused holder and a new holder to coexist.

Distinguish efficiency locks (duplicate work is acceptable) from correctness locks (corruption is not).

Correctness requires fencing tokens enforced by the protected resource, not by the lock service.

Renew short leases instead of setting long ones, and always verify ownership before releasing.

🧪 Practice

Give a concrete sequence where two clients hold the same lock, and show how a fencing token makes the outcome correct.

Write a Lua script for a safe Redis lock release that verifies ownership atomically.

Interview: Is a Redis lock safe for guarding a bank transfer? (Hint: ask what the storage layer does when a client with an expired lease writes.)

Quorum Systems

A quorum is a subset of nodes whose agreement is sufficient to act. Quorum systems are the mathematical backbone of consensus, replication, and partition handling: they turn "I cannot contact everyone" into "I can still proceed safely."

The property that makes them work is intersection. If every quorum overlaps with every other quorum, then any two operations that each obtained a quorum share at least one node — and that shared node carries the information linking them. Majority quorums are the simplest way to guarantee intersection, since two subsets each larger than half of a set must share a member.

 WHY MAJORITY INTERSECTION GUARANTEES SAFETY

  5 nodes: A B C D E

  write quorum: {A, B, C}          read quorum: {C, D, E}
                       \                       /
                        +---- C is in BOTH ---+

  C holds the write, so the read sees it. No arrangement of two majorities
  can avoid sharing at least one node -- which is why "majority" is not a
  heuristic but a proof.

def quorum_analysis(N):
    """Standard majority quorums."""
    majority = N // 2 + 1
    return {
        "nodes": N,
        "quorum_size": majority,
        "failures_tolerated": N - majority,
        "efficiency": majority / N,      # fraction that must respond
    }

for N in (3, 4, 5, 6, 7, 9):
    a = quorum_analysis(N)
    print(f"N={N}: quorum={a['quorum_size']} tolerates={a['failures_tolerated']}")

# N=3: quorum=2 tolerates=1      N=6: quorum=4 tolerates=2
# N=4: quorum=3 tolerates=1      N=7: quorum=4 tolerates=3
# N=5: quorum=3 tolerates=2      N=9: quorum=5 tolerates=4
#
# Even sizes gain nothing: 4 tolerates the same 1 failure as 3, while costing
# an extra node and an extra vote per decision. Always use odd cluster sizes.

Beyond simple majorities, several quorum designs trade properties differently:

Quorum type	Structure	Advantage
Majority	Any `N/2 + 1` nodes	Simple; maximum fault tolerance
Weighted	Nodes carry different vote counts	Reflects heterogeneous capacity
Grid	Nodes in a matrix; a row plus a column	Quorum size ~`2*sqrt(N)`
Hierarchical	Majority of majorities of subgroups	Rack and DC awareness
Flexible (Raft)	Election and replication quorums differ	Tune latency versus tolerance

Flexible quorums are the most practically interesting refinement. Consensus only requires that the election quorum intersects the replication quorum — not that each is a majority. That permits deliberately asymmetric configurations:

def flexible_paxos(N, replication_q, election_q):
    """Safety needs only Qr + Qe > N, not majorities on both sides."""
    return {
        "safe": replication_q + election_q > N,
        "write_latency_nodes": replication_q,
        "failures_tolerated_for_writes": N - replication_q,
    }

print(flexible_paxos(N=5, replication_q=2, election_q=4))
# {'safe': True, 'write_latency_nodes': 2, 'failures_tolerated_for_writes': 3}
#
# Writes need only 2 acknowledgements instead of 3 -- lower latency on the
# hot path -- at the cost of elections needing 4 nodes, which are rare.
# Optimizing the common case at the expense of the rare one.

Three practical considerations complete the picture.

Witness nodes. A lightweight node that votes but stores no data breaks ties without the cost of a full replica. A two-datacentre deployment with a witness in a third location is the standard way to get majority safety without a third full site.

Rack and zone awareness. A majority that happens to sit entirely in one rack provides no protection against that rack failing. Quorum placement must span failure domains, or the arithmetic is satisfied while the intent is not.

Read repair and hinted handoff. In Dynamo-style systems, quorum reads that observe divergent replicas repair them in the background, and writes destined for an unavailable node are held by a peer and delivered on recovery. These are what keep an eventually consistent quorum system converging in practice.

Key Takeaways

Quorums work because any two of them intersect, so information passes through the shared node.

Majority quorums tolerate N - (N/2 + 1) failures; even cluster sizes waste a node.

Flexible quorums need only Qr + Qe > N, allowing fast writes at the cost of more expensive elections.

Quorums must span failure domains, and witness nodes provide majority safety without a full third replica.

🧪 Practice

For N = 7, list three (Qr, Qe) pairs that are safe and rank them by write latency.

Explain why a 4-node cluster is strictly worse than a 3-node one.

Interview: You run two datacentres and want to survive losing either. What do you deploy? (Hint: a majority cannot exist within two equal groups — something must live somewhere else.)

Gossip Protocols

Gossip protocols spread information the way rumours spread through a crowd: each node periodically picks a few random peers and shares what it knows. There is no coordinator, no fixed topology, and no global view — yet information reaches every node quickly and the mechanism tolerates failures gracefully.

The motivation is that broadcast-based approaches do not scale. Having every node tell every other node costs O(N²) messages per round, and a coordinator that tells everyone is both a bottleneck and a single point of failure. Gossip achieves complete dissemination in O(log N) rounds with a constant number of messages per node per round, and it degrades smoothly when nodes fail.

 EPIDEMIC SPREAD: O(log N) ROUNDS

  round 0:  1 node knows          *
  round 1:  2 nodes               * *
  round 2:  4 nodes               * * * *
  round 3:  8 nodes               * * * * * * * *
  round 4:  16 nodes              ...

  10,000 nodes: ~14 rounds. At one round per second, the whole cluster
  knows within roughly 14 seconds -- with each node sending only a few
  messages per second, regardless of cluster size.

import random

class GossipNode:
    def __init__(self, node_id, peers, fanout=3):
        self.id, self.peers = node_id, peers
        self.fanout = fanout          # peers contacted per round; 3 is typical
        self.state = {}               # key -> (value, version, heartbeat)

    def tick(self):
        """One gossip round -- typically every 0.5 to 1 second."""
        targets = random.sample(self.peers, min(self.fanout, len(self.peers)))
        for peer in targets:
            # Push-pull: send our digest, receive theirs, exchange only the
            # differences. Push alone is slower to converge at the tail;
            # pull alone is slow to start. Push-pull is the usual choice.
            their_state = peer.exchange(self.digest())
            self.merge(their_state)

    def merge(self, other):
        for key, (value, version, hb) in other.items():
            mine = self.state.get(key)
            # Version comparison makes merge COMMUTATIVE and IDEMPOTENT, which
            # is what lets messages arrive in any order, repeatedly, without
            # corrupting the result.
            if mine is None or version > mine[1]:
                self.state[key] = (value, version, hb)

Gossip's properties come directly from its randomness:

Property	Consequence
Scalable	Constant work per node regardless of cluster size
Fault tolerant	No node is required; failures are routed around implicitly
Eventually consistent	Convergence is probabilistic, not immediate
Redundant	Nodes receive the same information several times
Self-healing	A recovering node is caught up by the next exchanges

The costs are equally direct. Convergence is eventual and probabilistic — there is no moment when the protocol declares completeness — and the redundancy means measurable steady-state bandwidth even when nothing changes. Tuning fanout and interval is a trade between convergence speed and network cost:

import math

def convergence_estimate(N, fanout, interval_s):
    rounds = math.log(N) / math.log(1 + fanout)    # approximate
    return {
        "rounds": round(rounds, 1),
        "seconds": round(rounds * interval_s, 1),
        "msgs_per_node_per_s": fanout / interval_s,
    }

print(convergence_estimate(N=1000, fanout=3, interval_s=1.0))
# {'rounds': 5.0, 'seconds': 5.0, 'msgs_per_node_per_s': 3.0}
print(convergence_estimate(N=1000, fanout=6, interval_s=1.0))
# {'rounds': 3.5, 'seconds': 3.5, 'msgs_per_node_per_s': 6.0}
#
# Doubling fanout doubles bandwidth and improves convergence by ~30%.
# Diminishing returns arrive quickly; fanout 3-5 is the usual sweet spot.

Gossip is used for exactly the problems where eventual convergence is acceptable and scale is high: cluster membership and failure detection (Cassandra, Consul, Serf), configuration propagation, aggregate metrics, and CRDT state exchange. It is emphatically not suitable where agreement is required — you cannot gossip your way to a leader election or a committed transaction, because gossip provides no decision point and no guarantee that everyone has converged yet.

Key Takeaways

Gossip disseminates information via random peer exchanges, reaching all nodes in O(log N) rounds with constant per-node cost.

It is highly fault tolerant and self-healing because no specific node is required for progress.

Merges must be commutative and idempotent, since messages arrive repeatedly and out of order.

It provides eventual convergence with no completion signal, so it cannot replace consensus for decisions requiring agreement.

🧪 Practice

Estimate the rounds and seconds to converge for 50,000 nodes at fanout 4 and a 500 ms interval.

Explain why the merge function must be idempotent, with a concrete failure if it is not.

Interview: Cassandra uses gossip for membership but Paxos for lightweight transactions. Why not gossip for both? (Hint: what does gossip never tell you about the state of the cluster?)

Membership and Failure Detection

Every distributed system must maintain a view of which nodes exist and which are alive. That sounds mechanical, and it is genuinely difficult, because FLP guarantees you can never be certain a node has failed — only that it has not responded within a time you chose.

Failure detectors are therefore characterized by two properties in tension:

Completeness: every genuinely failed node is eventually suspected.
Accuracy: no correct node is wrongly suspected.

You can have perfect completeness by suspecting everyone, and perfect accuracy by suspecting nobody. Every real detector picks a point between, and the timeout is the dial.

 THE DETECTION TRADE-OFF

  aggressive (timeout 500 ms)        conservative (timeout 30 s)

  + detects real failures fast       + almost never wrong
  - a GC pause looks like death      - a dead node keeps receiving traffic
  - false positives cause needless     for half a minute
    failovers, rebalances, and
    elections -- each of which
    costs availability

  The cost of a FALSE POSITIVE is usually higher than a slightly slower
  detection, because a spurious failover has all the cost of a real one.

A significant improvement over fixed timeouts is the phi accrual failure detector, which outputs a continuous suspicion level rather than a boolean. Instead of asking "is it dead?", it reports "how surprising is this silence, given the history of this node's heartbeat intervals?" — which lets different subsystems apply different thresholds to the same signal.

import math, statistics
from collections import deque

class PhiAccrualDetector:
    def __init__(self, window=1000, threshold=8.0):
        self.intervals = deque(maxlen=window)   # observed heartbeat gaps
        self.last_heartbeat = None
        self.threshold = threshold              # phi 8 ~ 1 in 10^8 chance of
                                                # a false positive

    def heartbeat(self, now):
        if self.last_heartbeat is not None:
            self.intervals.append(now - self.last_heartbeat)
        self.last_heartbeat = now

    def phi(self, now):
        """Suspicion level: -log10(probability this silence is normal)."""
        if not self.intervals or self.last_heartbeat is None:
            return 0.0
        elapsed = now - self.last_heartbeat
        mean = statistics.fmean(self.intervals)
        stdev = max(statistics.pstdev(self.intervals), 1e-3)
        # Probability that a normal interval would already have exceeded
        # `elapsed`. As elapsed grows past the historical distribution, phi
        # rises smoothly -- so a node on a slow network gets more slack
        # automatically, without any manual per-node tuning.
        p = 1 - _normal_cdf(elapsed, mean, stdev)
        return -math.log10(max(p, 1e-12))

    def is_suspected(self, now):
        return self.phi(now) > self.threshold

Membership protocols build on detection. The dominant design is SWIM (Scalable Weakly-consistent Infection-style Membership), used by Consul's Serf, HashiCorp's tooling, and others. Its two ideas both address false positives:

 SWIM: INDIRECT PROBING

  1. Node A probes node B directly.
  2. No response within the timeout.
  3. Instead of declaring B dead, A asks k random peers to probe B:

        A --ping-req(B)--> C --ping--> B
        A --ping-req(B)--> D --ping--> B

  4. If ANY of them reaches B, B is alive and the A-B path was the problem.
     Only if all indirect probes also fail is B suspected.

  This distinguishes "B is dead" from "A cannot reach B" -- an asymmetric
  network failure that a direct-probe-only detector would misdiagnose.

  SUSPICION MECHANISM: a suspected node is marked "suspect" and gossiped as
  such, giving it a window to refute the claim before being declared dead.

Approach	Detection quality	Message cost per round
All-to-all heartbeat	Fast, accurate	`O(N²)` — does not scale
Centralized monitor	Simple	`O(N)`, single point of failure
Gossip heartbeats	Good, scalable	`O(N)` total
SWIM	Very good; handles asymmetric failure	`O(1)` per node
Phi accrual	Adaptive per node	Depends on transport

Two operational points close the topic. First, suspicion should be gossiped before death is declared, so that a node experiencing a brief pause has a chance to refute it — the alternative is a cluster that churns membership every time a JVM collects garbage. Second, membership changes should be rate limited: a network blip that causes a large fraction of nodes to be simultaneously suspected must not trigger simultaneous rebalancing, which turns a transient event into a real outage.

Key Takeaways

Failure detection trades completeness against accuracy, and the cost of a false positive usually exceeds that of slower detection.

Phi accrual detectors output a continuous suspicion level derived from observed heartbeat history, adapting per node automatically.

SWIM's indirect probing distinguishes a dead node from an unreachable path, and its suspicion phase gives nodes a chance to refute.

Rate-limit membership changes so a transient network event does not trigger cluster-wide rebalancing.

🧪 Practice

Compute the message cost per round for all-to-all heartbeats versus SWIM at 1,000 nodes.

Explain how phi accrual gives a node on a slow link more slack without any manual configuration.

Interview: A brief network blip causes 40% of your cluster to be marked dead, and the resulting rebalance takes the service down. What do you change? (Hint: the detection was arguably correct — the problem is what happened next, and how fast.)

8.5 Distributed Transactions

A local transaction gets atomicity from a single storage engine that controls every write. Once the writes span two databases or two services, no such controller exists, and atomicity must be constructed from protocols — each with a different price.

Two-Phase Commit

Two-phase commit (2PC) is the classic protocol for making a transaction atomic across several participants. A coordinator drives two rounds: first it asks everyone whether they can commit, then it tells everyone the collective outcome.

The reason a second round is needed is that a participant cannot decide alone. If each simply committed when ready, one failing while others succeeded would leave the system half-updated. The prepare phase separates ability to commit from the decision to commit, so nobody acts until the outcome is known.

The pivotal obligation is what "prepared" means. A participant voting yes is making a binding promise: it has written everything it needs to durable storage, taken the necessary locks, and guarantees it can commit even if it crashes and restarts. It has given up the right to change its mind.

 TWO-PHASE COMMIT

 PHASE 1 (prepare)                    PHASE 2 (commit)
  coordinator                          coordinator
      |--- PREPARE ---> A                  |--- COMMIT ---> A
      |--- PREPARE ---> B                  |--- COMMIT ---> B
      |--- PREPARE ---> C                  |--- COMMIT ---> C
      |<--- YES ------- A                  |<--- ACK ------ A
      |<--- YES ------- B                  |<--- ACK ------ B
      |<--- YES ------- C                  |<--- ACK ------ C
      |                                    |
   all YES -> decide COMMIT             transaction complete
   any NO   -> decide ABORT

  Once a participant votes YES it has written a prepare record to disk and
  MUST be able to honour either outcome, across a crash and restart.

def two_phase_commit(coordinator, participants, txn):
    # PHASE 1: collect votes.
    votes = {}
    for p in participants:
        try:
            votes[p] = p.prepare(txn)      # p durably logs its intent here
        except (Timeout, ConnectionError):
            votes[p] = False               # unreachable counts as NO
    coordinator.log_decision(txn.id, "commit" if all(votes.values()) else "abort")
    # ^ THE decision point. This log write must be durable BEFORE phase 2,
    #   because it is the only record of what was decided.

    # PHASE 2: broadcast the outcome, retrying forever.
    decision = coordinator.read_decision(txn.id)
    for p in participants:
        # Participants cannot be allowed to give up. A participant that is
        # down must be told the outcome when it returns -- so this retries
        # indefinitely rather than failing.
        retry_until_success(lambda: p.commit(txn) if decision == "commit"
                                    else p.abort(txn))
    return decision

2PC guarantees atomicity, and its cost is severe: it is a blocking protocol. If the coordinator fails after participants have voted yes but before delivering the decision, those participants are stuck. They cannot commit (the decision might have been abort), cannot abort (it might have been commit), and must hold their locks until the coordinator returns.

 THE BLOCKING WINDOW

  coordinator: PREPARE ---> all vote YES ---> X CRASHES before deciding
  participant A: prepared, locks held, waiting ...
  participant B: prepared, locks held, waiting ...

  A and B cannot resolve this by talking to each other: neither knows what
  the coordinator decided, and the coordinator may have already told a third
  participant to commit. They must WAIT for the coordinator to recover.

  Meanwhile every row they locked is unavailable to everyone else. One
  coordinator failure can freeze a large part of the system.

Property	Behaviour
Atomicity	Guaranteed
Availability	Poor — blocks on coordinator failure
Latency	At least two round trips plus multiple disk syncs
Lock duration	Held across the whole protocol, including the network
Scalability	Degrades with participant count; failure probability compounds

The mitigations used in practice are worth knowing. Making the coordinator itself fault-tolerant — replicated via Raft, as Spanner and CockroachDB do — removes the single point of failure and turns 2PC from unusable into merely expensive. Participant timeouts with a cooperative termination protocol let participants ask each other what they know, which resolves some but not all blocking cases. And keeping the participant count small bounds both latency and the probability that any one participant fails mid-protocol.

2PC remains appropriate within a single trust and administrative domain where strict atomicity is required — across shards of one database, or in an XA transaction spanning a database and a message broker. Across microservice boundaries it is generally the wrong tool, because it couples the availability of every participant to every other, and sagas (two topics ahead) trade atomicity for that independence.

Key Takeaways

2PC separates the ability to commit from the decision, so no participant acts before the outcome is known.

A yes vote is a durable, binding promise the participant must honour across a crash.

It blocks: a coordinator failure after voting leaves participants holding locks with no way to resolve.

Replicating the coordinator via consensus is what makes 2PC viable; across service boundaries prefer sagas.

🧪 Practice

List every failure point in 2PC and state which are recoverable without the coordinator.

Compute the probability that at least one of eight participants fails during a 200 ms protocol, given a per-node failure rate of 0.1% per second.

Interview: Why is 2PC unsuitable across microservices? (Hint: consider what one service's availability now depends on, and for how long its rows are locked.)

Three-Phase Commit

Three-phase commit was designed to remove 2PC's blocking problem by inserting an extra round between the vote and the commit. The added phase communicates "we are all going to commit" before anyone actually does, so a participant that loses the coordinator can infer the outcome from its own state.

The insight is that 2PC blocks because a prepared participant cannot distinguish "the coordinator decided commit" from "the coordinator decided abort". 3PC removes that ambiguity by making the decision observable in an intermediate state: if a participant has received PRE-COMMIT, it knows every participant voted yes, so commit is the only possible outcome.

 THREE-PHASE COMMIT

  PHASE 1  CAN-COMMIT?  -> votes collected (no locks taken yet, in some
                           formulations)
  PHASE 2  PRE-COMMIT   -> "everyone voted yes; prepare to commit"
                           participants acknowledge and may now infer the
                           outcome on their own
  PHASE 3  DO-COMMIT    -> "commit now"

  RECOVERY RULE after coordinator failure:
    a participant in PRE-COMMIT state -> COMMIT   (everyone voted yes)
    a participant not yet in PRE-COMMIT -> ABORT  (nobody could have committed)

  Participants can also query each other and reach the same conclusion,
  which is what makes the protocol non-blocking.

def three_phase_commit(coordinator, participants, txn):
    # PHASE 1: can you commit?
    if not all(p.can_commit(txn) for p in participants):
        broadcast(participants, "abort", txn)
        return "aborted"

    # PHASE 2: pre-commit. This is the new phase. After it, every participant
    # KNOWS the vote was unanimous, so it can decide independently if the
    # coordinator vanishes.
    for p in participants:
        p.pre_commit(txn)                 # durably records "outcome is commit"

    # PHASE 3: do commit.
    for p in participants:
        p.do_commit(txn)
    return "committed"

def participant_recovery(self, txn):
    """Run when the coordinator is unreachable -- the point of 3PC."""
    if self.state(txn) == "pre-commit":
        return self.do_commit(txn)        # unanimous yes was established
    if self.state(txn) == "uncertain":
        peers = self.query_peers(txn)     # ask others what state they are in
        if any(s == "pre-commit" for s in peers):
            return self.do_commit(txn)    # someone got pre-commit -> commit
        return self.abort(txn)            # nobody did -> safe to abort
    return self.abort(txn)

The catch — and the reason 3PC is essentially absent from production systems — is that its non-blocking property depends on assuming a synchronous network with reliable failure detection. Under a network partition, that assumption fails catastrophically:

 WHY 3PC IS UNSAFE UNDER PARTITION

  Partition splits the cluster mid-protocol:

    side X: participants that RECEIVED pre-commit -> recovery rule says COMMIT
    side Y: participants that did NOT              -> recovery rule says ABORT

  Both sides follow the protocol correctly and reach OPPOSITE decisions.
  Atomicity -- the one thing the protocol exists to provide -- is violated.

  2PC blocks in this situation. 3PC produces a split-brain commit. Blocking
  is inconvenient; disagreeing is incorrect.

Aspect	2PC	3PC
Round trips	2	3
Blocks on coordinator failure	Yes	No (under synchrony)
Safe under partition	Yes (blocks)	No — can produce inconsistent commits
Latency	Lower	~50% higher
Production use	Widespread	Essentially none

The lesson generalizes well beyond this protocol, which is the real reason to study it: 3PC traded safety for liveness, and in distributed systems that trade is almost always wrong. A protocol that stalls can be resolved by an operator or by the coordinator returning. A protocol that commits inconsistently has produced corruption that no recovery procedure can undo, because there is no record of what the correct state should have been.

The modern resolution is neither 2PC nor 3PC in their classic forms, but 2PC with a consensus-replicated coordinator. Raft makes the coordinator's decision durable and available without assuming synchrony, which removes the blocking problem while preserving safety under partition. Spanner, CockroachDB, and YugabyteDB all take this approach.

Key Takeaways

3PC adds a pre-commit phase so participants can infer the outcome independently and avoid blocking.

Its non-blocking guarantee assumes a synchronous network and reliable failure detection.

Under a partition, the two sides can correctly follow the protocol and reach opposite decisions, violating atomicity.

The practical answer is 2PC with a consensus-replicated coordinator: safe under partition and non-blocking in practice.

🧪 Practice

Construct the partition scenario in which 3PC produces an inconsistent commit, naming each participant's state.

Explain why blocking is preferable to disagreement for a protocol whose purpose is atomicity.

Interview: How does a Raft-replicated coordinator fix 2PC's blocking without 3PC's unsafety? (Hint: what exactly was unavailable in the blocking case, and can that thing be made highly available on its own?)

Saga Pattern

A saga is a sequence of local transactions where each step commits independently, and failure is handled by running compensating transactions that semantically undo the completed steps. It is the standard answer to multi-service transactions once 2PC's coupling has been ruled out.

The motivation is availability. 2PC holds locks across services for the duration of the protocol and requires every participant to be reachable. A saga holds no distributed locks at all — each step commits locally and immediately, so services remain independently available. What you surrender is isolation: intermediate states are visible to everyone.

 SAGA: FORWARD PATH AND COMPENSATION

  T1: reserve inventory   -> OK
  T2: charge card         -> OK
  T3: create shipment     -> FAILS
                             |
        compensate in REVERSE order:
  C2: refund card         <- undo T2
  C1: release inventory   <- undo T1

  Note what compensation is NOT: it is not a rollback. The charge and the
  refund BOTH appear on the customer's statement. The system reaches a
  semantically equivalent state, not an identical one.

Two coordination styles exist, matching the orchestration-versus-choreography distinction from Chapter 7:

Style	Structure	Strength	Weakness
Orchestration	A central coordinator drives steps	State is explicit and queryable	The coordinator is a component to run
Choreography	Each service reacts to events	No central component	The process exists nowhere; hard to debug

# Orchestrated saga with explicit compensation, persisted between steps.
class Saga:
    def __init__(self, saga_id, store):
        self.id, self.store, self.steps = saga_id, store, []

    def step(self, name, action, compensation):
        self.steps.append((name, action, compensation))
        return self

    def execute(self, context):
        completed = []
        for name, action, compensation in self.steps:
            try:
                # Persist BEFORE and AFTER each step. If the orchestrator
                # crashes, recovery reads this log to learn where it was and
                # whether the last step's outcome is known.
                self.store.mark(self.id, name, "started")
                result = action(context)
                context.update(result)
                self.store.mark(self.id, name, "completed")
                completed.append((name, compensation))
            except Exception as exc:
                self.store.mark(self.id, name, "failed", error=str(exc))
                self._compensate(completed, context)
                raise SagaFailed(name, exc)
        return context

    def _compensate(self, completed, context):
        for name, compensation in reversed(completed):
            try:
                compensation(context)
                self.store.mark(self.id, name, "compensated")
            except Exception as exc:
                # A failed compensation cannot be resolved automatically --
                # the system is now in a state no code path intended.
                self.store.mark(self.id, name, "compensation_failed",
                                error=str(exc))
                alert_oncall("saga_compensation_failed", saga=self.id, step=name)
                # Continue compensating the rest: partial cleanup beats none.

order_saga = (Saga(saga_id, store)
    .step("inventory", reserve_inventory,  release_inventory)
    .step("payment",   charge_card,        refund_card)
    .step("shipping",  create_shipment,    cancel_shipment)
    .step("notify",    send_confirmation,  send_cancellation))

Sagas satisfy ACD but not I: atomicity in the semantic sense (all steps complete or all are compensated), consistency eventually, durability per step — but no isolation. That missing I produces real anomalies you must design around:

Anomaly	What happens	Countermeasure
Dirty read	Another transaction reads an intermediate saga state	Semantic lock / status flag
Lost update	A concurrent write overwrites a saga's write	Version checks / optimistic concurrency
Non-repeatable read	The same read returns different values mid-saga	Re-read and re-validate

# Semantic lock: mark the entity as in-flight so others can see and respect it.
def reserve_inventory(ctx):
    db.execute("""
        UPDATE inventory
           SET available = available - :qty,
               reserved  = reserved  + :qty,
               saga_id   = :saga_id        -- the semantic lock: visible to
         WHERE sku = :sku AND available >= :qty   -- everyone, enforced by
    """, ctx)                                     -- application convention
    if db.rowcount == 0:
        raise InsufficientStock()
    # Other readers see `reserved` and can choose to treat the item as
    # pending rather than available. This is isolation implemented in the
    # data model rather than by the database.

Three design rules make sagas workable in practice. Order steps so that irreversible actions come last — send the email after the shipment is confirmed, not before — since a step with no compensation cannot be undone. Make every step and compensation idempotent, because retries are guaranteed under at-least-once messaging. And persist saga state, so a crashed orchestrator resumes rather than restarting or, worse, leaving the saga half-run with nobody tracking it.

Key Takeaways

A saga replaces distributed atomicity with a sequence of local transactions plus compensations, keeping services independently available.

It provides atomicity, consistency, and durability but no isolation, so intermediate states are visible.

Countermeasures for missing isolation — semantic locks, version checks — are implemented in the data model, not by the database.

Order irreversible steps last, make everything idempotent, and persist saga state so a crash resumes rather than restarts.

🧪 Practice

Design the saga for a hotel-plus-flight booking, listing every compensation and identifying which step must come last.

Show a concrete dirty read in a saga and how a semantic lock prevents the resulting bug.

Interview: A saga has completed four of six steps and the orchestrator dies. What must be true for recovery to be correct? (Hint: recovery must determine whether step four actually succeeded — where does it look?)

Compensating Transactions

A compensating transaction is the operation that semantically undoes a completed step. It is easy to describe and consistently underestimated, because the natural mental model — "just roll it back" — does not survive contact with the real world.

The critical distinction is between rollback and compensation. A rollback erases history: the database discards the uncommitted change and no observer ever saw it. A compensation is a new forward transaction that counteracts an earlier committed one. Both remain in the record, both were visible to anyone looking, and both may have triggered side effects of their own.

 ROLLBACK vs COMPENSATION

  ROLLBACK (local transaction)        COMPENSATION (saga)

  BEGIN                                T1: balance 100 -> 50   [COMMITTED]
    balance 100 -> 50                      (visible to everyone; a statement
  ROLLBACK                                  was generated; a webhook fired)
  -> balance is 100. No trace.         C1: balance 50 -> 100   [COMMITTED]
                                       -> balance is 100, and the ledger
                                          shows a debit AND a credit.

Compensations fall into three categories, and knowing which one you are dealing with determines how much design work is required:

Category	Example	Compensation
Fully reversible	Reserve inventory	Release the reservation — state restored
Semantically reversible	Charge a card	Refund — net zero, but both entries persist
Irreversible	Send an email, launch a rocket	None possible; only mitigation

# Fully reversible: the compensation restores the prior state exactly.
def reserve_seat(ctx):
    db.execute("UPDATE seats SET status='reserved', hold_id=:hold "
               "WHERE id=:seat AND status='available'", ctx)

def release_seat(ctx):
    # Guarded so it only releases OUR hold. Without the hold_id check, a
    # late compensation could release a seat someone else has since taken.
    db.execute("UPDATE seats SET status='available', hold_id=NULL "
               "WHERE id=:seat AND hold_id=:hold", ctx)

# Semantically reversible: a new transaction with its own identity.
def refund_payment(ctx):
    return payments.refund(
        charge_id=ctx["charge_id"],
        amount=ctx["amount"],
        # Idempotency is mandatory: compensations are retried like everything
        # else, and a double refund is as bad as a double charge.
        idempotency_key=f"refund-{ctx['saga_id']}-{ctx['charge_id']}",
    )

# Irreversible: cannot be undone -- only followed up.
def compensate_notification(ctx):
    # You cannot unsend an email. The only options are to send a correction
    # or to have ordered this step later so it never ran.
    send_email(ctx["user"], template="order_cancelled_after_confirmation")

Four properties every compensation must have:

Idempotent. It will be retried. A refund issued twice is a defect of the same severity as the original failure.

Commutative where possible. Compensations may execute in an order you did not plan, particularly when several run concurrently after a multi-branch failure.

Always eventually succeed. A compensation that can fail permanently leaves the system inconsistent with no automated path back. Compensations should be simple enough that failure is nearly impossible — and where it is not, they must escalate to a human rather than silently give up.

Ordered in reverse. Undo in the opposite order of doing, so each compensation runs against the state its corresponding action created.

# The hardest case: what to do when compensation itself fails.
def compensate_with_escalation(step, ctx, max_attempts=5):
    for attempt in range(max_attempts):
        try:
            step.compensation(ctx)
            return "compensated"
        except TransientError:
            time.sleep(backoff(attempt))
        except PermanentError as exc:
            break
    # Automated recovery is exhausted. The system is in an intermediate state
    # that no code path intended, so record it precisely and hand it to a
    # human -- silently continuing would hide a real inconsistency.
    ledger.record_inconsistency(
        saga=ctx["saga_id"], step=step.name, context=ctx,
        requires="manual_reconciliation")
    alert_oncall("compensation_failed_permanently", saga=ctx["saga_id"])
    return "manual_intervention_required"

The most valuable design habit is to ask, for every step you add to a workflow, "what undoes this?" — before writing the step. Steps that turn out to have no compensation should be moved as late as possible, replaced with a reversible two-phase equivalent (reserve then confirm, rather than commit immediately), or made conditional on everything else having already succeeded.

Key Takeaways

Compensation is a new forward transaction, not a rollback: both the action and its undo remain visible in the record.

Categorize each step as fully reversible, semantically reversible, or irreversible, and design accordingly.

Compensations must be idempotent, order-tolerant, and effectively guaranteed to succeed.

Design the compensation before the action; move irreversible steps last or restructure them as reserve-then-confirm.

🧪 Practice

Categorize and write compensations for: reserve inventory, charge card, generate an invoice PDF, send an SMS, provision a VM.

Explain why release_seat checks hold_id and what breaks without it.

Interview: How do you handle a step that genuinely cannot be compensated? (Hint: two structural answers — one changes when it runs, one changes what it does.)

Outbox Pattern

The outbox pattern solves the most common partial-failure bug in event-driven systems: a service must update its database and publish an event, and these are two different systems that cannot share a transaction.

The problem is unavoidable without it. Whichever order you choose, there is a window where a crash leaves the two out of sync — a database row with no event, or an event describing a change that was never persisted.

 THE DUAL-WRITE PROBLEM (no correct ordering exists)

  Option A: write DB, then publish
      db.insert(order)        -> OK
      X crash
      broker.publish(event)   -> never happens
      => the order exists and no downstream service knows

  Option B: publish, then write DB
      broker.publish(event)   -> OK
      X crash
      db.insert(order)        -> never happens
      => downstream services process an order that does not exist

  Wrapping them in try/catch does not help: the crash can occur between any
  two instructions, including inside the catch block.

The outbox pattern makes the two writes atomic by putting the event in the same database as the business data, in the same transaction. A separate process then reads the outbox table and publishes to the broker.

 OUTBOX PATTERN

  +----------------- ONE DATABASE TRANSACTION -----------------+
  |  INSERT INTO orders  (...)                                 |
  |  INSERT INTO outbox  (event_type, payload, ...)            |
  +------------------------------------------------------------+
                            |
                            |  committed atomically: both or neither
                            v
              +------------------------------+
              |  relay / CDC process         |  polls or tails the WAL
              +------------------------------+
                            |
                            v  at-least-once publish
                      [ message broker ]
                            |
                            v
                      consumers (must be idempotent)

CREATE TABLE outbox (
    id             BIGSERIAL PRIMARY KEY,
    aggregate_type TEXT        NOT NULL,     -- 'order'
    aggregate_id   TEXT        NOT NULL,     -- for partition keying
    event_type     TEXT        NOT NULL,     -- 'OrderPlaced'
    payload        JSONB       NOT NULL,
    created_at     TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    published_at   TIMESTAMPTZ                -- NULL until published
);
-- Partial index keeps the scan small even when the table holds history.
CREATE INDEX idx_outbox_unpublished ON outbox (id) WHERE published_at IS NULL;

def place_order(order):
    with db.transaction():                 # ONE transaction, TWO tables
        order_id = db.execute(
            "INSERT INTO orders (...) VALUES (...) RETURNING id", order)
        db.execute("""
            INSERT INTO outbox (aggregate_type, aggregate_id, event_type, payload)
            VALUES ('order', :id, 'OrderPlaced', :payload)
        """, {"id": order_id, "payload": json.dumps(event_for(order, order_id))})
    return order_id
    # A crash anywhere leaves a consistent state: either both rows exist or
    # neither does. The event is now guaranteed to be published eventually.

def relay_loop():
    while True:
        rows = db.execute("""
            SELECT id, aggregate_id, event_type, payload FROM outbox
             WHERE published_at IS NULL
             ORDER BY id                       -- preserves per-aggregate order
             LIMIT 100
             FOR UPDATE SKIP LOCKED            -- several relays can run safely
        """)
        for row in rows:
            broker.publish(row.event_type, row.payload, key=row.aggregate_id)
            db.execute("UPDATE outbox SET published_at = NOW() WHERE id = :id",
                       row)
            # A crash between publish and UPDATE republishes on restart.
            # That is AT-LEAST-ONCE by design -- consumers must be idempotent.
        if not rows:
            time.sleep(0.2)

There are two ways to run the relay, and the choice matters:

Approach	Mechanism	Trade-offs
Polling relay	Query the outbox table on a loop	Simple, portable; adds query load and latency
CDC (log tailing)	Read the database WAL (e.g. Debezium)	Near-zero latency and no query load; more infrastructure

The pattern's guarantee is precisely at-least-once with atomicity, not exactly-once. A crash between publishing and marking the row published causes a duplicate, which is why the outbox does not remove the need for idempotent consumers — it removes the need for lost events, which is the harder problem.

Two operational details close it out. The outbox table must be pruned, or it becomes the largest table in the database; deleting published rows older than a retention window is a routine job. And ordering is preserved only per aggregate: publishing with aggregate_id as the partition key gives per-entity ordering downstream, which — as Chapter 7 established — is the ordering guarantee that actually matters.

The mirror-image pattern, the inbox, applies the same idea on the consuming side: record the processed message ID in the same transaction as the business change, giving exactly-once effect end to end.

Key Takeaways

The dual-write problem has no correct ordering; a crash between the two writes always leaves an inconsistency.

The outbox makes them atomic by writing the event into the same database transaction as the business data.

A relay publishes from the outbox with at-least-once delivery, so consumers must still be idempotent.

Use CDC for low latency, poll for simplicity; prune the table and key publishes by aggregate ID to preserve per-entity order.

🧪 Practice

Explain why try/catch around the publish call cannot fix the dual-write problem.

Write the pruning job and decide the retention window, justifying it against consumer replay needs.

Interview: The outbox guarantees the event is published. Why do consumers still need to be idempotent? (Hint: look at the two statements in the relay loop and ask what a crash between them causes.)

Conflict Resolution and Last-Write-Wins

When a system allows concurrent writes to the same data on different replicas — which every AP system does — conflicting versions will occur. Conflict resolution is the policy for deciding what the converged value should be, and it is where "eventually consistent" stops being free.

Last-write-wins (LWW) is the simplest policy: attach a timestamp to every write and keep the one with the highest value. It requires no application logic, no extra metadata beyond a timestamp, and produces a single deterministic answer on every replica.

It also silently discards data, and it does so for a reason that has nothing to do with which write was more important: physical clock skew, as established in Section 8.3.

 LWW LOSES A WRITE, WITH NO ERROR ANYWHERE

  real time    node A (300 ms fast)      node B (100 ms slow)
  10:00:00.20  write "Alice" -> ts .500
  10:00:00.40                            write "Bob" -> ts .100

  Converged value: "Alice", because .500 > .100.

  Bob's write happened LATER and is discarded. No conflict is reported, no
  log records it, and Bob simply observes that his change "did not save".
  The system is behaving exactly as configured.

The alternatives, in order of increasing effort and decreasing data loss:

Strategy	Determinism	Data loss	Application effort
Last-write-wins	Yes	Silent	None
Highest version wins	Yes	Explicit	Version tracking
Sibling values	N/A	None	Client must resolve
Application merge	Yes	None if correct	Per-type logic
CRDTs	Yes	None	Choose the right type

# LWW with a tiebreak, which is the minimum bar for correctness.
def lww_resolve(v1, v2):
    if v1.timestamp != v2.timestamp:
        return v1 if v1.timestamp > v2.timestamp else v2
    # Equal timestamps are COMMON at millisecond resolution under load.
    # Without a deterministic tiebreak, different replicas can pick
    # differently and never converge at all -- a worse failure than losing
    # a write. Any total order works; every replica must use the same one.
    return v1 if v1.node_id > v2.node_id else v2

# Version-based: at least the loss is detectable.
def version_resolve(v1, v2):
    rel = compare_vector_clocks(v1.clock, v2.clock)
    if rel == "before":     return v2
    if rel == "after":      return v1
    return None             # concurrent -> a REAL conflict; do not guess

# Application merge: use domain knowledge the database does not have.
def merge_shopping_cart(v1, v2):
    # Union of additions, intersection of removals: nothing a customer added
    # is ever lost, which is the behaviour the business actually wants.
    merged = {}
    for item_id in set(v1.items) | set(v2.items):
        q1 = v1.items.get(item_id, 0)
        q2 = v2.items.get(item_id, 0)
        merged[item_id] = max(q1, q2)      # deliberate: prefer keeping items
    return Cart(items=merged)

The decision rule is short and worth applying explicitly:

Use LWW when the value is a single overwritable field whose history does not matter (a cached display name, a last-seen timestamp, a sensor reading), losing an occasional write is acceptable, and clocks are reasonably synchronized.

Do not use LWW when the value accumulates (carts, sets, counters), when every write carries business meaning (financial entries, audit records), or when users will notice their change vanishing.

# Cassandra applies LWW per COLUMN, which is more forgiving than per row --
# and is a distinction worth knowing.
#
#   node A writes: name = 'Alice'   (ts 100)
#   node B writes: email = 'b@x.com' (ts 90)
#
# Both survive: they are different columns, so there is no conflict at all.
# Only concurrent writes to the SAME column contend. This is why Cassandra's
# LWW is less lossy in practice than a naive whole-document LWW -- but it
# also means a "row" can be assembled from writes that never coexisted.

That last observation generalizes: LWW at a fine granularity loses less data but can produce a combined state that no single writer ever intended. Choosing the granularity is part of choosing the policy.

Key Takeaways

LWW is deterministic and free, and it silently discards writes based on clock skew rather than on importance.

Equal timestamps require a deterministic tiebreak, or replicas will not converge at all.

Version-based resolution makes conflicts detectable so they can be handled deliberately.

Reserve LWW for overwritable single fields; use merges or CRDTs for anything that accumulates or carries business meaning.

🧪 Practice

Construct a clock-skew scenario where LWW loses a critical write, and give two ways to prevent it.

Explain why column-level LWW is less lossy than document-level, and what strange state it can produce instead.

Interview: Your users report that profile edits occasionally revert. The database uses LWW. Walk through the diagnosis. (Hint: two writes and two clocks — which comparison decided the outcome?)

CRDTs

Conflict-free replicated data types are data structures designed so that concurrent updates cannot conflict. Rather than detecting conflicts and resolving them, CRDTs make the merge operation mathematically guaranteed to produce the same result on every replica, in any order, however many times it is applied.

The guarantee comes from algebra. If the merge function is commutative (order does not matter), associative (grouping does not matter), and idempotent (repetition does not matter), then replicas receiving the same set of updates converge regardless of the order or duplication with which they arrive. Those three properties define a join-semilattice, and every CRDT is one.

 WHY THE ALGEBRA GUARANTEES CONVERGENCE

  Replica A receives: u1, u2, u3
  Replica B receives: u3, u1, u2, u1     (different order, one duplicate)

  commutative -> order does not matter
  associative -> grouping does not matter
  idempotent  -> the repeated u1 changes nothing

  Therefore A and B hold the same state. No coordination occurred, no
  conflict was detected, and none needed to be.

There are two families. State-based (CvRDT) replicas exchange their full state and merge with a join function; simple and robust to message loss, but the messages are large. Operation-based (CmRDT) replicas broadcast operations, which must be delivered exactly once but are small. Most production systems use a hybrid, exchanging deltas.

class GCounter:
    """Grow-only counter: the foundational CRDT."""
    def __init__(self, node_id, nodes):
        self.node_id = node_id
        self.counts = {n: 0 for n in nodes}   # each node owns ONE entry

    def increment(self, n=1):
        self.counts[self.node_id] += n        # only ever touch your own

    def value(self):
        return sum(self.counts.values())

    def merge(self, other):
        # Elementwise max. Commutative, associative, idempotent -- because
        # each node's entry only ever grows, max never loses an increment.
        for n, c in other.counts.items():
            self.counts[n] = max(self.counts.get(n, 0), c)


class PNCounter:
    """Increment and decrement, built from two G-Counters."""
    def __init__(self, node_id, nodes):
        self.inc = GCounter(node_id, nodes)
        self.dec = GCounter(node_id, nodes)   # decrements counted separately,
                                              # because a single counter that
    def increment(self, n=1): self.inc.increment(n)   # went both ways would
    def decrement(self, n=1): self.dec.increment(n)   # break monotonicity
    def value(self):  return self.inc.value() - self.dec.value()
    def merge(self, other):
        self.inc.merge(other.inc)
        self.dec.merge(other.dec)


class LWWElementSet:
    """A set where add and remove each carry a timestamp."""
    def __init__(self):
        self.adds, self.removes = {}, {}      # element -> latest timestamp

    def add(self, e, ts):    self.adds[e]    = max(self.adds.get(e, 0), ts)
    def remove(self, e, ts): self.removes[e] = max(self.removes.get(e, 0), ts)

    def contains(self, e):
        # Bias matters: >= means add wins a tie, > would mean remove wins.
        # Choose deliberately -- for a shopping cart, add-wins is right.
        return e in self.adds and self.adds[e] >= self.removes.get(e, 0)

    def merge(self, other):
        for e, ts in other.adds.items():
            self.adds[e] = max(self.adds.get(e, 0), ts)
        for e, ts in other.removes.items():
            self.removes[e] = max(self.removes.get(e, 0), ts)

CRDT	Purpose	Notable characteristic
G-Counter	Increment-only counter	Per-node entries merged by max
PN-Counter	Increment and decrement	Two G-Counters
G-Set	Grow-only set	Union; removal impossible
2P-Set	Add and remove once	Removed elements can never return
OR-Set	Add and remove freely	Unique tags per add; add-wins semantics
LWW-Register	Single value	Timestamp-based, with LWW's caveats
RGA / Logoot	Ordered sequence	The basis of collaborative text editing

The cost of CRDTs is metadata and semantics. Tombstones for removed elements must be retained so a late-arriving add does not resurrect them, so a set that sees heavy churn grows even as its logical size stays constant. And the convergence is guaranteed but its meaning is fixed by the type: an OR-Set is add-wins, so a concurrent add and remove yields presence, which is right for a cart and wrong for a permission revocation.

# The semantic point, stated concretely.
# OR-Set is ADD-WINS. Concurrent add and remove -> the element is PRESENT.
#
#   shopping cart:  add-wins is correct  (never lose a customer's item)
#   access control: add-wins is DANGEROUS (a concurrent revoke loses to a
#                   concurrent grant, and the user keeps access)
#
# For revocation you want remove-wins, or you should not be using an
# eventually consistent structure for that decision at all.

CRDTs are used in Redis (CRDB), Riak, Automerge and Yjs (collaborative editing), Figma's multiplayer engine, and Apple's Notes sync. They are the right tool when you need writes to be available everywhere with no coordination and the data type fits one of the known structures. They are the wrong tool when the operation requires a global invariant — "never oversell inventory" cannot be a CRDT, because enforcing it requires exactly the coordination CRDTs exist to avoid.

Key Takeaways

CRDTs make merges commutative, associative, and idempotent, so replicas converge without coordination or conflict detection.

State-based CRDTs ship full state and tolerate message loss; operation-based ship small operations and need reliable delivery.

Costs are metadata growth (tombstones) and fixed semantics — an OR-Set is add-wins whether or not that is what you want.

They cannot enforce global invariants; anything requiring a cross-replica constraint still needs consensus.

🧪 Practice

Prove that G-Counter's merge is commutative, associative, and idempotent.

Implement an OR-Set with unique tags and show why 2P-Set cannot re-add a removed element.

Interview: Can you build a bank account balance as a PN-Counter? (Hint: the arithmetic converges perfectly — what happens to the rule that the balance must never go below zero?)

← System Design