The Trouble with Distributed Systems

Everything that can go wrong in a distributed system, in one chapter. The book assumes the polite kind of fault: a node crashes, a packet is lost, a process stalls. Nobody lies on purpose. Even with that assumption, almost every intuition from single-machine programming turns out to be wrong.

Vocabulary

Terms I met in this chapter, defined at the end.

partial failurenon-Byzantine faultnetwork partitiontimeoutunbounded delaybackpressurejitterPhi Accrual failure detectorcascading failurecircuit switchingpacket switchingbursty trafficQoSadmission controltime-of-day clockmonotonic clockNTPPTPsmearingclock driftconfidence intervalTrueTimelast write winscausalitylogical clockversion vectorMVCCprocess pauseleasesplit brainquorumzombieSTONITHfencing tokenoptimistic concurrency controlcompare-and-swap

Faults and partial failures

On a single machine, things either work or they do not. A bad instruction, a kernel panic, a full disk: the failure is total, and the machine tells you. A supercomputer handles failure the same way, by letting any fault escalate into a full stop and restarting the whole job from a checkpoint.

A distributed system cannot do that. Some nodes are fine while others are broken, and the fine ones cannot tell which is which. This is a partial failure, and it is nondeterministic: the same operation succeeds, fails, or hangs depending on timing you do not control. Stopping everything for repair is not acceptable when the system is serving users, so it has to keep working with some of its parts missing.

The upside is real. If the system survives failed nodes, you can restart one node at a time for a rolling upgrade, or kill a slow virtual machine and ask for a new one. The price is that every fault, including unlikely ones, has to be thought through and tested for. Suspicion pays off.

Unreliable networks

Machines in a shared-nothing system have exactly one way to talk to each other: the network. Send a request and wait for a reply. The figure shows three ways that goes wrong; the chapter lists five, and from the sender's side they are all the same thing.

1. The request was lost.
2. The request is waiting in a queue and will be delivered later.
3. The remote node has crashed.
4. The remote process is paused, for example in a long garbage collection, and will resume later.
5. The remote node processed the request, but the response was lost or delayed.

No response tells you nothing about which one happened. The only tool is a timeout: stop waiting and assume the reply is not coming. Note what the timeout does not tell you. It does not say whether the remote side did the work.

Network faults in practice

Partitions, where one part of the network is cut off from the rest, are not rare. Public clouds have frequent transient glitches. Well-run private datacenters are calmer, but switches get misconfigured, cables get pulled, and sharks do bite undersea cables. The chapter's advice is blunt: if the error handling for a network fault is not defined and tested, anything can happen, up to a cluster that deadlocks permanently or deletes its own data. Showing users an error while the network is broken is a perfectly valid design.

Detecting faults

Many pieces of a system need to notice when a node has died. A load balancer must take it out of rotation. In single-leader replication a follower must be promoted. Sometimes you get a hint: the OS refuses a connection with a RST or FIN when nothing is listening on the port, a router answers with ICMP unreachable, or a supervisor script announces the crash to the others, as HBase does. None of this is something you can count on. TCP acknowledging a packet does not mean the application processed it. If you need to know a request succeeded, you need a reply from the application itself, and in general you have to be ready to get no reply at all.

Timeouts and unbounded delays

How long should the timeout be? There is no good answer. A long timeout means a long wait before a dead node is declared dead. A short timeout means declaring nodes dead that were only slow.

Declaring a live node dead is dangerous twice over. If it was in the middle of some work, the work may now happen twice. And its load moves to the remaining nodes. If the node was slow because the system was overloaded, you have just made the overload worse, and in the extreme every node declares every other node dead. That is a cascading failure.

If the network guaranteed delivery within d and a healthy node always replied within r, a timeout of 2d + r would be safe. Real systems give neither bound. Delays are unbounded because of queueing, in four places.

• Several nodes send to the same destination at once, so the switch queues the packets. If the queue fills, packets are dropped and resent, on a network that is working fine.
• A packet arrives at a machine whose CPUs are all busy, so the OS queues it until the application gets a turn.
• In a virtualised environment the whole OS is paused for tens of milliseconds while another VM uses the core, and the hypervisor buffers the incoming data meanwhile.
• TCP flow control, also called backpressure, queues data at the sender before it even enters the network. TCP also retransmits what it believes is lost, after a timeout derived from observed round trips. The application does not see the loss; it sees the delay.

UDP is the other end of the trade: no flow control, no retransmission, and so fewer sources of variable delay, at the price of losing data. It is the right choice when late data is worthless, as in a voice call.

In a public cloud, resources are shared with customers you cannot see. A batch job next door can saturate a link. So rather than a fixed timeout, measure response times and their variability (jitter) and adjust. The Phi Accrual failure detector, used in Akka and Cassandra, does exactly this. TCP's retransmission timer works the same way.

Synchronous versus asynchronous networks

A phone call reserves a fixed slice of bandwidth along the whole route for its duration. Nothing queues, so the end-to-end delay is bounded. That is circuit switching, and it is possible because a call sends a constant number of bits per second.

Ethernet and IP are packet switched. They are built for bursty traffic: a web page, an email, a file transfer have no bandwidth requirement, they just want to finish as fast as possible, and sit idle in between. Packet switching gives a burst the whole wire and nothing when idle. The cost is queueing, and so unbounded delay. ATM tried to be both. With quality of service and admission control you can approximate a bounded delay on a packet network, but datacenter networks and the internet do not do this.

The same trade-off shows up in CPUs. Time-sharing a core between threads means a thread sometimes waits in the run queue, but it uses the hardware far better than a static allocation would. Variable delay is not a law of nature. It is what you get when you choose utilisation over predictability, and that choice is why the wire is cheap.

Unreliable clocks

Durations and points in time are both needed everywhere: has this request timed out, when does this cache entry expire, which of these two writes came first, when did this error happen. Each machine measures time with its own quartz oscillator, which runs a little fast or a little slow, so each machine has its own idea of the time. NTP narrows the gap by asking a server with a better clock. It cannot close it.

Monotonic versus time-of-day clocks

Modern computers expose two clocks, and they answer different questions.

• Time-of-day clock. "What time is it?" Seconds since an epoch. Synchronised by NTP, which means it can jump backwards when NTP corrects it. SystemTime in Rust, clock_gettime(CLOCK_REALTIME) in C.
• Monotonic clock. "How much time has passed?" A counter that only moves forward. Its absolute value is meaningless, often nanoseconds since boot, so comparing readings from two machines means nothing. Instant in Rust, CLOCK_MONOTONIC in C.

Use the monotonic clock for timeouts and durations. It needs no synchronisation and does not care about small inaccuracy. On a multi-socket server there may be a timer per CPU and the OS has to hide the difference, so even this clock is not perfect, but it is the right one.

Clock synchronisation and accuracy

Time-of-day clocks need an external source, and the ways that goes wrong are many.

• Quartz drifts, by something like 200 parts per million, which is 17 seconds a day if left alone.
• A clock too far from the NTP server gets stepped rather than slewed, so applications see time jump.
• A node firewalled off from NTP drifts silently. This happens in practice.
• Leap seconds have crashed many large systems. The least bad fix is to make NTP servers lie and spread the extra second over a day. That is smearing.
• In a VM the hardware clock is virtualised, and when the VM is paused for another tenant the clock appears to jump forward.
• On devices you do not control, the clock is whatever the user set. Some set it years off on purpose.

Much better accuracy is possible when it matters. European MiFID II regulation requires high-frequency trading systems to be within 100 microseconds of UTC, which is achieved with GPS receivers and the Precision Time Protocol. It takes expertise and monitoring. A misconfigured NTP daemon or a blocked port quietly throws the whole thing away.

Relying on synchronised clocks

Three things about clocks that software routinely assumes away: a day does not always have 86,400 seconds, a time-of-day clock can go backwards, and the time on one node can be very different from the time on another.

Robust software treats clocks like it treats the network: fine most of the time, and wrong occasionally, with the wrongness handled. The nasty part is that a bad clock goes unnoticed. A broken CPU or NIC fails loudly. A broken quartz or a misconfigured NTP client lets everything seem fine while the clock wanders away from reality, and software that depends on it produces silent, subtle data loss rather than a crash. If you depend on synchronised clocks, monitor the offsets and evict any node whose clock drifts too far, the same way you evict a dead node.

Timestamps for ordering events

The tempting use of a clock is to order writes across nodes. Which of these two writes is more recent?

In the figure, client A writes to node 1 and client B increments the same value on node 3 after seeing A's write. Node 3's clock is slightly behind, so B's write gets an earlier timestamp than A's. Last write wins keeps the "newer" one by timestamp and throws B's increment away. The write that causally came second is lost, with no error.

Last write wins is the conflict resolution in Cassandra, Riak and most multi-leader setups. Timestamps from the client instead of the server do not help. The problems are structural:

• A node with a lagging clock cannot overwrite anything written by a node with a fast clock until the skew has passed. Writes disappear silently.
• LWW cannot tell a write that happened after another from one that was truly concurrent. The second needs a merge; the first needs ordering. A timestamp gives neither reliably.
• Two nodes can produce the same millisecond timestamp. A random tiebreaker resolves it and can also violate causality.

Even with tight NTP, a packet can be sent at 100 ms by the sender's clock and arrive at 99 ms by the receiver's. The fix is to stop using physical time for ordering. Logical clocks, which are counters that advance on events rather than oscillating crystals, and version vectors track causality directly. Physical clocks measure elapsed time; that is what they are for.

Clock readings have a confidence interval

A clock reading is not a point. It is a range, and the honest API returns the range. Google Spanner's TrueTime and Amazon's ClockBound return [earliest, latest]. The width depends on how long since the last sync and how fast the local quartz may have drifted since.

Synchronised clocks for global snapshots

MVCC gives a long read-only transaction a consistent snapshot while small transactions keep writing. On one machine the snapshot is a transaction counter. Across many machines, handing out globally ordered transaction IDs is a bottleneck, so Spanner uses TrueTime intervals as IDs. If interval A ends before interval B starts, A definitely happened before B. If they overlap, Spanner waits out the uncertainty before committing. That makes commit latency depend on clock accuracy, which is why Google runs GPS and atomic clocks in every datacenter. This only works because the clock API admits how unsure it is.

Process pauses

A lease is a lock with an expiry. If the holder crashes, nobody has to release it; it simply runs out.

Take a database with one leader per shard. Only the leader accepts writes. The leader holds a lease from the other nodes and renews it before it expires. To know whether it is still leader, it checks the lease against the clock:

while (true) {
  request = getIncomingRequest();
  if (lease.expiryTimeMillis - System.currentTimeMillis() < 10000) {
    lease = lease.renew();
  }
  if (lease.isValid()) {
    process(request);
  }
}

Two bugs. The expiry was set by another machine's clock and is being compared to this one's. And the thread can stop between isValid() and process() for far longer than the lease lasts. If it stops for 15 seconds, another node is leader by the time process() runs, and the old one has no idea.

Threads stop for longer than you think, for reasons you cannot see from the code:

• Stop-the-world garbage collection, which can take minutes on a large heap.
• A virtual machine suspended and resumed, for example for live migration.
• A laptop lid closing. An OS context switch under load. A hypervisor steal.
• Synchronous disk I/O, including class loading on the JVM, which hits the disk at surprising moments.
• Swapping. A page fault can block a thread on disk for as long as the disk likes, and thrashing makes it worse.
• SIGSTOP, sent by an operator or a tool that forgot to send SIGCONT.

On a single machine, mutexes, semaphores, atomic counters, lock-free structures and blocking queues let you reason about this. None of them exist across a network. A node in a distributed system has to assume it can be paused for any length of time at any point, during which the rest of the world keeps moving, and that it will not be told when it resumes.

Real-time systems exist and provide hard bounds on this, at the cost of specialised OSes, real-time garbage collectors, certified languages and much lower throughput. For most server software the practical mitigations are smaller: treat a GC pause like a planned outage and drain the node first, or restart processes before the heap gets big enough to need a full collection.

Knowledge, truth and lies

A node knows only what it has received over the network. It cannot see whether another node is alive, only that a message did or did not arrive in time. The system's notion of truth therefore cannot live on any one node.

Majority rules

Decisions are made by a quorum: a vote among the nodes, usually a strict majority. A node that has been declared dead by the majority is dead, however alive it feels. It cannot overrule the vote, because one node's view is exactly what is unreliable. Consensus algorithms are built on this, and the next chapter is about them.

Distributed locks and leases

A lease is the right tool whenever there must be exactly one of something: one leader per shard to avoid split brain, one client updating a resource at a time, one worker processing a given input file. The failure mode is that two nodes believe they hold the lease at once.

Client 1 holds the lease, pauses, and the lease expires. Client 2 is granted it and writes. Client 1 wakes up believing it still holds the lease and writes too. Split brain, and a corrupted file. The pause is the story in this figure, but a network delay produces the same picture: client 1's write leaves on time and arrives after client 2's.

Fencing off zombies

A zombie is a former leaseholder that has not learned it lost the lease and keeps acting on it. You cannot prevent zombies. A paused process is, by definition, not running the code that would notice. What you can do is make sure a zombie cannot do damage.

The crude way is STONITH, shoot the other node in the head: cut the zombie off the network, kill its VM. That handles a paused process. It does nothing about the delayed request in the second figure, which is already on the wire.

The robust way is a fencing token.

Every time the lock service grants a lease it also returns a number that increases with every grant. Every write to the storage service carries the client's current token. Storage remembers the highest token it has seen and rejects anything lower. Client 1 writes with token 33 after client 2 has already written with 34; the write is refused, however alive client 1 feels and however late the packet is. The check moves from the client, which cannot be trusted to know its state, to the resource, which can.

The token has many names. Chubby calls it a sequencer, Kafka an epoch, ZooKeeper exposes the zxid or a node's cversion, etcd uses the revision with the lease ID, Hazelcast's FencedLock generates one explicitly. In Paxos it is the ballot number and in Raft the term. Fencing is final: a rejected write stays rejected. That is the difference from optimistic concurrency control, where a rejected write is retried against the new state.

Fencing with multiple replicas

Fencing works because the storage is a single point that can keep the highest token. If the storage is itself replicated, each replica has to enforce the token, and a write has to be accepted by a quorum of them before it counts. Otherwise a zombie can find one replica that has not seen the new token and write there. The token has to travel with every write, to every replica, and the replicas have to agree on what the highest one is. At that point you are building consensus, which is the next chapter.

The big idea

Nothing a node observes on its own can be trusted: not the absence of a reply, not its clock, not its own belief that it is still running without interruption. A correct system therefore never lets one node's local view decide anything that matters. It puts the decision where it can be checked, in a quorum or in the resource itself with a fencing token, and it designs every timeout and retry to be safe when the guess is wrong.

Concepts

• Partial failure. Some parts of the system are broken and the rest cannot tell which. Nondeterministic, and the defining property of distributed systems.
• Timeout. The only fault detector available, and it conflates lost, slow, paused and dead. Pick it by measurement; write code that is correct when it misfires.
• Two clocks. Time-of-day for "when", synchronised and allowed to jump. Monotonic for "how long", never compared across machines.
• Clock readings are intervals. TrueTime and ClockBound say so explicitly. Every other clock API is hiding the same uncertainty.
• Process pause. A thread can stop anywhere, for any length of time, and resume without being told. Code that reads the clock then acts on the result has a window in it.
• Lease. A lock with an expiry, so a crash does not hold it forever. A lease holder cannot safely act on its own belief that the lease is valid.
• Fencing token. A monotonically increasing number issued with each lease and checked by the resource. Turns "I think I hold the lock" into "the storage knows who holds it".
• Quorum. Truth is a majority vote, because no single node's view is reliable.

Trade-offs

• Packet switching versus circuit switching. Unbounded delay in exchange for much higher utilisation, which is why the network is cheap and the timeout is hard.
• TCP versus UDP. Reliability in exchange for delay variance. Choose UDP only where late data is useless.
• Long versus short timeouts. Slow failover versus false positives that can cascade into an outage.
• Physical versus logical clocks. Physical clocks give wall time and elapsed time; logical clocks give causal order. Using one for the other's job is the source of most clock bugs.
• STONITH versus fencing tokens. STONITH is simple and stops a paused node; only a token stops a delayed request.
• Fencing versus optimistic concurrency control. A fenced write is dead; an OCC write is retried. Fencing when the losing writer must not win, OCC when it just needs the latest state.
• Real-time guarantees versus throughput. Bounded pauses are available, at the cost of specialised runtimes and far less work per second.

In the wild

I came to this chapter from barista-rs, the fair-share dispatcher I am building, which hands jobs to workers and has to survive every failure listed above. Two things changed because of it.

One clock. The first design had dispatcher replicas writing last_heartbeat from their own clocks and the reaper on another replica comparing it to its own. Forty seconds of skew between two replicas reads as forty seconds of silence, and with a thirty-second lease timeout that is a live worker declared dead. Now every lease timestamp is written and compared with the database's now(). One clock measures both ends, so there is no skew to be wrong about. It is still a wall clock, so the lease timeout has to be much larger than any step the database host might see, but that is one clock to monitor instead of N.

A fencing token the database enforces. The lease and the storage are the same Postgres row, so there is no separate lock service and no separate storage to teach about tokens. The token is the job's attempt, incremented on every dispatch and echoed by the worker in every message. Each write from a worker is an UPDATE ... WHERE attempt = $1, so a worker that pauses, loses its lease and wakes up finds its writes change zero rows. The side effects, releasing capacity, adjusting the tenant's tag, run only when that update reports one row. The one deliberate difference from the book: the row accepts a token that is equal to the current attempt, not greater-or-equal, because a job has exactly one live attempt. And since fencing only protects the database, the worker also kills its own container after 25 seconds without contact, strictly less than the 30-second lease, so the zombie stops burning CPU before the job is handed to anyone else.

The grid at the bottom of that diagram is the chapter's failure list with barista-rs's answer to each. The one in red, the database crashing, is out of scope and needs a standby.

I did not reach for etcd or ZooKeeper, and this chapter made clear when I would have to: if the truth lived in dispatcher memory and the database were only a log, I would need one elected leader; if jobs were sharded across databases with no shared transaction, I would need a coordinator outside them; if several services needed locks and configuration, one lock service beats five sets of tables. None of those hold today. A single Postgres row is lease server, storage and fencing check in one place, and that is what makes the simple version correct.

Takeaways

• A missing reply is not a crash. It is one of five things, and the code has to be correct for all of them.
• Timeouts are measured, not chosen, and declaring a node dead has a cost on the survivors.
• Read durations from the monotonic clock. Never order events across machines by a time-of-day clock.
• A clock reading is an interval. An API that returns a point is rounding.
• A thread can pause anywhere for any length of time. Any check-then-act on a lease has a hole in it.
• A lease without a fencing token is a race. The resource, not the holder, decides who holds the lock.
• Single-node primitives do not cross the network. What replaces them is a quorum.

Vocabulary, defined

• Partial failure. A fault in some parts of a system that the other parts cannot reliably detect.
• Non-Byzantine fault. A crash, loss or delay. Nodes may be slow or dead, but never malicious or lying.
• Network partition. A fault that splits the network so that groups of nodes cannot reach each other.
• Timeout. Giving up on a reply after a chosen wait, and treating the absence as failure.
• Unbounded delay. A delivery time with no upper limit, the result of queueing in packet-switched networks.
• Backpressure. Flow control that makes a sender slow down when the receiver cannot keep up. In TCP it queues data at the sender.
• Jitter. The variability of response times, as opposed to their average.
• Phi Accrual failure detector. A detector that estimates the probability a node is down from the observed distribution of heartbeat delays, rather than a fixed timeout.
• Cascading failure. One failure increasing load or suspicion on the survivors until they fail too.
• Circuit switching. Reserving a fixed path and bandwidth for a connection, giving bounded delay. Telephone networks.
• Packet switching. Sending independent packets that share links opportunistically, giving high utilisation and queueing. Ethernet, IP.
• Bursty traffic. Load that arrives in short bursts with idle gaps, which packet switching serves well.
• QoS, admission control. Prioritising and scheduling packets, and rate-limiting senders, to approximate bounded delay on a packet network.
• Time-of-day clock. Wall-clock time since an epoch. Synchronised externally and allowed to jump.
• Monotonic clock. A forward-only counter for measuring elapsed time. Meaningless across machines.
• NTP. Network Time Protocol, which adjusts a machine's clock from a server with a better one, over the network.
• PTP. Precision Time Protocol, hardware-assisted synchronisation to microsecond accuracy, often with a GPS source.
• Smearing. Spreading a leap second across many hours so no clock ever sees a 61-second minute.
• Clock drift. The gradual divergence of a quartz clock from true time, in parts per million.
• Confidence interval. The [earliest, latest] range a clock reading actually represents.
• TrueTime. Spanner's clock API, which returns that interval and lets commits wait out the uncertainty.
• Last write wins. Resolving concurrent writes by keeping the one with the latest timestamp and discarding the rest.
• Causality. The happens-before relation: B saw A, so B must be ordered after A.
• Logical clock. A counter that advances on events, used to order them by causality rather than by physical time.
• Version vector. One counter per replica, carried with each value, used to tell concurrent writes from causally ordered ones.
• MVCC. Multiversion concurrency control: readers see a consistent snapshot while writers create new versions.
• Process pause. A thread stopping for an unbounded time without the code knowing: GC, VM suspension, swapping, signals.
• Lease. A lock with an expiry time, so a crashed holder releases it by doing nothing.
• Split brain. Two nodes each believing they are the single leader or lock holder.
• Quorum. A majority of nodes agreeing, used as the system's definition of truth.
• Zombie. A node that has lost its lease and does not know it, still acting as the holder.
• STONITH. Shoot the other node in the head: forcibly cutting off or killing a suspected zombie.
• Fencing token. An increasing number issued with each lease and checked by the resource, so stale holders' writes are rejected.
• Optimistic concurrency control. Writing without a lock, then checking nothing changed underneath; retrying if it did.
• Compare-and-swap. An atomic operation that updates a value only if it still equals an expected value. The primitive under OCC and under fenced writes.