08

The trouble with distributed systems

On one machine, things either work or crash cleanly. Across a network, some parts fail while others carry on, and nobody can be sure which ones. Messages vanish or turn up late, clocks drift and jump, and a process can freeze mid-thought and wake up acting on stale beliefs. This chapter is a tour of these faults. Each model below lets you cause one, so you can see why a node can't trust its own view of the world, and why systems rely on majorities, fencing tokens and explicit system models instead.

Unreliable networks: no response means nothing

partial-failureunreliable-network
Node A sends a request and waits. Pick what the network or the remote node does to it, and compare what really happened with what A can observe.

Node A sends a request to node B

t = 0 ms
A (sender)B (remote)idlerequest →← responsetimeout 70 ms

Try: flip to the sender's view and run "request lost", "remote crashed" and "response lost"; A sees exactly the same thing. Then run "response lost" with retry on and count how many times B executed the request.

A's verdict
idle
times B executed it
0
original only
what really happened
The remote node DID process the request, but the response was dropped on the way back.
The only failure detector is a timeout, but packet delay has no upper bound. Most of the variation comes from queueing at switches, plus retransmits and remote pauses. Pick a timeout and see how many healthy nodes it would wrongly declare dead.

Round-trip times to a perfectly healthy node (4,000 requests)

0100200300400500600+round-trip time (ms)requests (log)+1 beyond 600 msp50p99p99.9timeout

Green responses beat the timeout; red ones came from a node that was fine but got declared dead. Try: set a tight 50 ms timeout, then raise the load to 90%. Queueing alone moves the whole tail, and no timeout is both fast and safe.

false 'dead' verdicts
1.90%
76 of 4000 healthy replies were late
false alarms / hour
1,368
5 nodes pinging each other every second
time to detect a real crash
≥ 100 ms
users wait (or see errors) this long
p50 / p99 / p99.9
12 / 227 / 450 ms
worst seen 684 ms; no upper bound exists
If the network guaranteed a maximum delay d and nodes always handled requests within r, the timeout 2d + r would never be wrong. Telephone circuits come close by reserving bandwidth. Datacenter networks give that up for burst throughput, so the best you can do is measure the distribution and adapt, as phi accrual detectors in Akka and Cassandra do.

Unreliable clocks: time-of-day vs monotonic

time-of-day-clockmonotonic-clockclock-drift
A quartz clock drifts, and NTP periodically steps it back toward the server's time, which can move it backwards. Measure an operation with both kinds of clock.

Time-of-day clock error vs true time

measured op-50510050100true time (s)clock − true (ms)true time

Each NTP sync yanks the time-of-day clock to the server's time, possibly backwards (amber markers). The monotonic clock is never stepped. Try: choose "stepped back 1.5 s" and time the 1-second operation starting at 59.5 s. Then pick "NTP unreachable" and watch the error grow without bound.

true duration
1,000.0 ms
monotonic clock says
1,000.20 ms
off by 0.20 ms (drift only)
time-of-day clock says
991.37 ms
includes an NTP step of -8.8 ms
worst offset in 2 min
13.7 ms
2 backward jumps; 200 ppm ≈ 17.3 s/day

Ordering events by timestamp

lww-clock-skewclock-confidence-interval
Timestamps from different machines are only as trustworthy as their clocks' synchronization. Last-write-wins replication quietly loses data under a few milliseconds of skew (the book's Figure 8-3). Spanner is the exception: it treats every clock reading as an interval and waits out the uncertainty.

Client B increments x after reading A's write (it is causally later)

node 1clock exactnode 2clock exactnode 3clock -3 msA: set x=1 ts 42.0000true t = 0.0 msB: read x=1B: set x=2 ts 41.9990true t = 2.0 msx=1x=2 droppedx=2 droppedtrue time →

Every replica keeps the value with the greatest timestamp. Try: with node 3 just 3 ms slow, B's increment carries a smaller timestamp than the write it read, so every node silently discards it. Slide the skew positive, then switch to logical clocks.

A's timestamp
42.0000
B's timestamp
41.9990
smaller, though B happened later
final x on all replicas
1
increment silently lost, no error

Clock readings as confidence intervals

txn Atxn Bs_A = 8.0s_B = 2.0051015A commits (true t=0)

Shaded bars are each clock's [earliest, latest] interval; the dashed tick is what the clock reads, the solid tick the timestamp chosen. Try: with no commit wait and B starting 2 ms later, B gets a smaller timestamp and its snapshot misses A. Turn on commit wait: A sits out the uncertainty before acknowledging, so any later transaction must get a larger timestamp. Shrinking ε shrinks the wait.

s_A vs s_B
8.0 vs 2.0
B's snapshot sees A?
no
A committed first, yet B reads before it
latency cost
0.0 ms
≈ 2ε added to every read-write commit

Process pauses, leases and fencing tokens

process-pauselock-leasefencing-token
Client 1 holds a lease on a lock and is writing a file when a garbage-collection pause freezes it. The lease expires and client 2 takes over. Then client 1 wakes up, still believing it holds the lock (Figures 8-4 and 8-5).

A lease-based lock protecting a file in storage

lock serviceclient 1client 2storagelease → client 1 (token 33)lease → client 2 (token 34)stop-the-world GCacquireok, token 33writeacquireok, token 34writewriteacceptGC pause 45lease 33 expiredwakes, thinks lease still validacceptaccept → CORRUPT

Client 1 checked its lease, then froze. Try: a 45 s pause without fencing, then turn fencing on. Lower the pause below about 15 s and the lease is still valid when it wakes. A pause between roughly 15 and 30 s is the scary middle: client 1 writes without a lease and nothing notices, because client 2 has not written yet.

outcome
data corrupted
client 1's stale write clobbered client 2's
client 1 wakes at
t=61
its lease expired at t=35, it doesn't know
writes accepted
3/3
C1:33✓ C2:34✓ C1:33✓
Checking the lease right before writing doesn't help: the pause can land between the check and the write. Only the resource can safely reject a stale holder, which is why the token has to be checked by the storage service, not the client.

The truth is defined by the majority

truth-by-majoritynetsplit
Five nodes exchange heartbeats. A node that misses heartbeats is suspected, and a leader is replaced only when a majority agrees. Inject a fault on A and see who gets to decide.

Five nodes, heartbeats and leader elections

tick 0
A♛ leader · epoch 1Bfollows A · e1Cfollows A · e1Dfollows A · e1Efollows A · e1

A freezes at t=10 and wakes at t=34 still believing it leads. Its epoch-1 writes are refused. Green links work both ways, amber arrows only one way, faint red not at all. Try: the GC pause, then step the tick slider around 34–36 to watch A act as a zombie leader for two ticks.

would a write commit now?
A (epoch 1)5/5 acks ✓
event log
  • nothing yet

Byzantine faults: when nodes lie

byzantine-fault
Crashed nodes are silent. Byzantine nodes can tell each quorum a different story. Count how many nodes it takes for any two quorums to share at least one honest member.

Two quorums that must agree

quorum 1 (3 nodes) hears "commit"quorum 2 (3 nodes) hears "abort"1☠34

A protocol must make progress after hearing from n − f nodes, because f may never answer. Any two such quorums overlap in n − 2f nodes, and at least one of those must be honest to carry the truth across. Try: f = 1 with crash faults needs 3 nodes; switch to Byzantine and n = 3 breaks (the lone overlap node tells each side what it wants), n = 4 works.

quorum size n − f
3
honest nodes in overlap
1
overlap 2 − 1 liars
verdict
quorums agree
needs n ≥ 4 (3f + 1)

System models, safety and liveness

system-modelsafety-liveness
To reason about correctness, write down what may go wrong: a timing model and a node-failure model. Then state which properties must always hold (safety) and which only need to hold eventually (liveness).

Pick a system model

Crash-stopCrash-recoveryByzantine
Synchronous
Partially synchronous
Asynchronous
Partially synchronous × Crash-recovery
behaves synchronously most of the time, but bounds are sometimes exceeded arbitrarily; nodes crash and come back; durable storage survives, memory doesn't.
The most useful model for real datacenter systems. Safety always; liveness during well-behaved periods. Nodes must persist votes and logs to disk so a restart doesn't make them forget promises.
ZooKeeper, etcd, Raft, Spanner (Paxos)

Try: compare partially synchronous × crash-recovery (what real systems assume) with asynchronous × crash-stop (FLP).

Safety or liveness?

0/7
  • Uniqueness: no two nodes ever hold the same lock lease at once.
  • Monotonic sequence: a later lock grant always carries a larger fencing token.
  • Availability: a node that asks for the lock and doesn't crash eventually gets an answer.
  • Replicas converge to the same value once writes stop.
  • A write acknowledged as committed is never lost.
  • A read never returns a value that no one wrote.
  • Every request eventually receives a response.

Safety = nothing bad happens (a violation happens at a point in time and can't be undone). Liveness = something good eventually happens. Algorithms must keep safety in every model; liveness is allowed to depend on the network eventually behaving.

  • takeawayA missing response tells you nothing: the request may have been lost, delayed or executed. Timeouts are the only detector, they are always a guess, and retries need idempotence.
  • takeawayClocks drift, jump and disagree. Use monotonic clocks for durations, don't order cross-node events by wall-clock timestamps (LWW drops writes), and if you must, use explicit uncertainty intervals as Spanner does.
  • takeawayA node can be paused or cut off at any moment without knowing it, so no single node decides. Use quorums for the truth and fencing tokens at the resource, and design for a partially synchronous, crash-recovery system model: safety always, liveness eventually.

Key concepts · 18

book chapter 8
  1. Partial failure

    Some parts of the system are broken in unpredictable ways while others keep working.

    A single computer is designed to either work or crash cleanly; a distributed system is nondeterministic, and you may not even know whether an operation succeeded. Fault tolerance has to be designed in from the start.

    ↑ see it in the visualization
  2. Asynchronous packet network

    A network that gives no guarantee when, or whether, a packet will arrive.

    When a request gets no response, it could have been lost, queued, the remote crashed or paused, or the response was lost. The sender cannot tell these apart, and a retry may execute the operation twice.

    EthernetIPTCP
    ↑ see it in the visualization
  3. Timeouts as failure detection

    Declaring a node dead when it hasn't responded within a fixed period.

    A short timeout detects crashes quickly but falsely kills slow nodes, whose load moves to others and can cascade. A long timeout leaves users waiting. There is no correct value, only a trade-off.

    ↑ see it in the visualization
  4. Unbounded delays and network congestion

    Variable packet delay caused mostly by queueing: at switches, in the OS, at a busy CPU, under TCP flow control and retransmission.

    Datacenter networks and the internet are packet-switched to use bandwidth efficiently, which means queues instead of reserved capacity. When load rises, the tail of the delay distribution grows without a hard limit.

    TCP retransmissionnoisy neighbours in the cloud
    ↑ see it in the visualization
  5. Adaptive timeouts (phi accrual)

    Choosing timeouts from the measured response-time distribution and its jitter instead of a constant.

    Systems that observe round-trip times can raise their suspicion level gradually, which balances detection delay against false positives as conditions change.

    AkkaCassandra
    ↑ see it in the visualization
  6. Time-of-day clock

    A wall-clock reading (e.g. seconds since the epoch) kept in sync with NTP.

    When NTP finds it too far off it can jump the clock backwards, so subtracting two readings can give a negative duration. It's fine for showing a date but wrong for measuring elapsed time.

    clock_gettime(CLOCK_REALTIME)System.currentTimeMillis()
    ↑ see it in the visualization
  7. Monotonic clock

    A clock guaranteed to move only forward, used to measure durations such as timeouts.

    Its absolute value means nothing and can't be compared across machines, but the difference between two readings on one machine is a reliable elapsed time.

    clock_gettime(CLOCK_MONOTONIC)System.nanoTime()
    ↑ see it in the visualization
  8. Clock drift and NTP synchronization

    Quartz clocks run fast or slow depending on temperature; NTP corrects them over the network, only as accurately as the network delay allows.

    Google assumes 200 ppm of drift: about 6 ms between syncs every 30 s, or 17 s a day. A firewalled NTP server, a misconfiguration or a VM pause can leave a clock badly wrong without anyone noticing.

    ↑ see it in the visualization
  9. Last write wins under clock skew

    Resolving conflicts by keeping the write with the greatest wall-clock timestamp.

    A node whose clock is a few ms slow writes timestamps smaller than writes it has already seen, so its causally later writes are silently dropped. Logical clocks order events by causality instead.

    CassandraRiak (LWW mode)
    ↑ see it in the visualization
  10. Clock confidence intervals (TrueTime)

    Treating a clock reading as a range [earliest, latest] with an explicit uncertainty bound.

    If two intervals don't overlap, the order of events is known. Spanner waits out the uncertainty before committing, which is why Google puts GPS receivers and atomic clocks in every datacenter to keep it around 7 ms.

    Google SpannerAWS ClockBound
    ↑ see it in the visualization
  11. Process pauses

    A thread stopped for an arbitrary time without noticing: stop-the-world GC, VM suspension, context switches, disk I/O, swapping, SIGSTOP.

    Code that checks a condition and then acts can be frozen between the two steps. When it wakes up, the world, and its lease, may have moved on.

    JVM GClive VM migration
    ↑ see it in the visualization
  12. Lease

    A lock with a timeout: the holder must renew it before it expires, or someone else may take over.

    Leases keep a crashed holder from blocking everyone forever, but a paused holder doesn't know its lease has run out and can act on stale authority.

    ChubbyZooKeeperetcd
    ↑ see it in the visualization
  13. Fencing token

    A number that increases every time the lock is granted, sent with every write to the protected resource.

    The storage service rejects any write carrying a token older than one it has already seen, so a zombie client woken from a pause can't corrupt data. The resource itself has to check.

    ZooKeeper zxid / cversion
    ↑ see it in the visualization
  14. The truth is defined by the majority

    Decisions such as declaring a node dead or choosing a leader are made by a quorum vote, not by one node.

    A node can't trust its own view: it may be the one cut off, or the one that just paused. If a majority declares it dead, it must step down, even if it feels fine.

    ↑ see it in the visualization
  15. Network partition (netsplit)

    Part of the network is cut off from the rest, while each side keeps working.

    Only the side holding a majority can keep electing leaders and committing writes. Partitions are surprisingly common, even inside one datacenter, so software must handle them deliberately.

    ↑ see it in the visualization
  16. Byzantine faults

    Nodes that behave arbitrarily: they send corrupted or conflicting messages, or lie on purpose.

    Tolerating f such nodes needs n ≥ 3f + 1, compared with 2f + 1 for crash faults. It matters for aerospace and blockchains. In a datacenter, nodes are usually assumed honest, but checksums and input validation still guard against weak forms of lying.

    PBFTBitcoinflight computers
    ↑ see it in the visualization
  17. System model

    An abstraction of the faults an algorithm must tolerate: timing (synchronous, partially synchronous, asynchronous) × node failures (crash-stop, crash-recovery, Byzantine).

    An algorithm is only correct relative to a model. Partially synchronous with crash-recovery is the most realistic for most systems.

    ↑ see it in the visualization
  18. Safety and liveness

    Safety: nothing bad ever happens. Liveness: something good eventually happens.

    A safety violation happens at a specific moment and can't be undone, so it must hold in every situation. Liveness may come with caveats, such as only once the network has recovered.

    ↑ see it in the visualization