The Trouble with Distributed Systems

Chapter 1: Unreliable Networks

The internet and most datacenter networks are asynchronous packet networks. When you send a message, the network gives you absolutely no guarantee about when (or whether) it will arrive. Here is the complete list of things that can go wrong when node A sends a request to node B:

From A's perspective, all five scenarios look identical: A sent a request and did not receive a response. That's it. There is no way for A to distinguish between them.

Network Partitions

A network partition (or "netsplit") is when the network splits into two or more groups of nodes that can communicate within the group but not across groups. All nodes are alive and processing requests. They just can't see each other.

This is different from a node crashing. In a crash, the failed node stops doing things. In a partition, both sides keep operating — potentially making conflicting decisions, accepting conflicting writes, electing conflicting leaders.

Partitions are not rare. A 2011 study of Microsoft datacenters found that network failures caused about 5 link failures per day, each affecting tens of thousands of devices. Google's internal data shows similar numbers. In cloud environments, partial partitions (where some links fail but others work) are especially insidious.

Simulate the Failure Modes

The simulation below shows node A sending a request to node B. Use the buttons to inject different failure modes and observe that A always sees the same thing: silence.

Timeouts: The Only Tool

Since you can't distinguish failure modes, the only option is to set a timeout: wait for some duration, and if no response arrives, assume something went wrong. But what should the timeout be?

There is no universally correct timeout. It depends on your network, your workload, and how bad false positives are compared to slow detection. We'll explore this tradeoff deeply in Chapter 2.

Chapter 2: Detecting Failures

You're running a cluster of five database replicas. One of them stops responding. Is it dead? Is it just slow? Is the network between you and it broken while it's still serving other clients? You need to decide — because if it's dead, you need to redirect traffic and maybe elect a new leader. If it's just slow, declaring it dead could cause a cascade of unnecessary failovers.

This is the failure detection problem. And your only tool is timeouts.

The Timeout Dilemma: Worked Example

Let's make this concrete. Suppose your service sends heartbeats every 1 second. Normally, responses arrive within 5ms. You need to decide: how many missed heartbeats constitute a failure?

Where Delays Come From

Network delays are not random noise — they have specific, mechanical causes. Understanding these helps you reason about reasonable timeout values.

Notice that most of these are queueing problems. Packets, threads, and processes all queue. The more loaded the system, the longer the queues, the higher the latency. This is why the 99th percentile of response time is far worse than the median.

The Phi Accrual Failure Detector

Rather than a fixed timeout, the phi accrual failure detector (used by Akka, Cassandra) adapts based on observed response times. Instead of a binary "alive or dead," it outputs a suspicion level (φ) that increases over time since the last heartbeat.

The beauty: if the network gets noisier (higher σ), the detector automatically becomes more lenient. If the network is rock-solid (low σ), it detects failures faster.

Timeout Tuning Simulator

The simulation below shows a node sending heartbeats. The response time varies randomly. Use the timeout slider to see the tradeoff: too short and you get false positives (declaring a healthy node dead), too long and you're slow to detect a real failure.

Python: Simple Phi Accrual Detector

python
import math
import time
from collections import deque

class PhiAccrualDetector:
    """Phi accrual failure detector (simplified Cassandra-style)."""

    def __init__(self, threshold=8, window_size=100):
        self.threshold = threshold       # phi above this = declared dead
        self.intervals = deque(maxlen=window_size)
        self.last_heartbeat = None

    def heartbeat(self):
        """Called when a heartbeat arrives."""
        now = time.monotonic()
        if self.last_heartbeat is not None:
            interval = (now - self.last_heartbeat) * 1000  # ms
            self.intervals.append(interval)
        self.last_heartbeat = now

    def phi(self):
        """Current suspicion level. Higher = more likely dead."""
        if self.last_heartbeat is None or len(self.intervals) < 2:
            return 0.0

        now = time.monotonic()
        elapsed = (now - self.last_heartbeat) * 1000  # ms since last heartbeat

        # Compute mean and variance of intervals
        mean = sum(self.intervals) / len(self.intervals)
        variance = sum((x - mean) ** 2 for x in self.intervals) / len(self.intervals)
        std = math.sqrt(variance) if variance > 0 else 0.1

        # CDF of normal distribution: P(X <= elapsed)
        z = (elapsed - mean) / std
        cdf = 0.5 * (1 + math.erf(z / math.sqrt(2)))

        # phi = -log10(1 - cdf), clamped to avoid log(0)
        p_late = max(1e-15, 1.0 - cdf)
        return -math.log10(p_late)

    def is_alive(self):
        return self.phi() < self.threshold

# Usage:
detector = PhiAccrualDetector(threshold=8)
# In your heartbeat loop:
# detector.heartbeat()  — call when each heartbeat arrives
# detector.is_alive()   — check if the node is still alive

Chapter 3: Unreliable Clocks

Every computer has a quartz crystal oscillator that ticks at a nominal frequency. But "nominal" is doing a lot of work in that sentence. Quartz crystals drift. A typical crystal oscillates at 32.768 kHz, but temperature, voltage, and manufacturing variance cause it to be slightly fast or slow. A drift of 35 parts per million (ppm) — common for consumer-grade crystals — means the clock gains or loses about 3 seconds per day.

On a single machine, this doesn't matter much. The OS periodically adjusts the clock using NTP (Network Time Protocol) and everything works fine. But in a distributed system, each node has its own clock, and they all drift independently. You cannot assume that "now" on node A is the same as "now" on node B.

Two Types of Clocks

Operating systems expose two fundamentally different clocks. Confusing them causes bugs that are maddeningly difficult to diagnose.

Why Timestamps Can't Order Events

Consider two database nodes using last-write-wins (LWW) conflict resolution. Each write is stamped with the node's local wall-clock time. The write with the highest timestamp wins.

Here's what happens when clocks are skewed:

This is not a theoretical concern. It is one of the most common sources of silent data loss in distributed databases that use LWW (including some default configurations of Cassandra).

Clock Drift Simulator

The simulation below shows three nodes with independent clock drift. NTP corrections occur periodically but introduce jumps. Watch how two events that happen in a clear temporal order get reversed timestamps.

Worked Example: How Much Can Clocks Drift?

Chapter 4: Clock Synchronization and Confidence

NTP gives you a timestamp. But how accurate is it? The terrifying answer: you don't know. NTP measures the round-trip time to a time server and estimates one-way delay as half the round-trip. But if the network is asymmetric (as it often is), this estimate is wrong. NTP gives you a time, but no error bars.

Google decided this was unacceptable. For their globally distributed database Spanner, they needed to know not just "what time is it?" but "what is the earliest and latest time it could be right now?"

Google's TrueTime

Google's TrueTime API uses a combination of GPS receivers and atomic clocks in every datacenter. Instead of returning a single timestamp, it returns a confidence interval:

Using Confidence Intervals for Ordering

With confidence intervals, you can determine when two events are definitely ordered versus when their order is ambiguous.

Spanner uses this to implement external consistency (a stronger guarantee than serializability). When a transaction commits, Spanner deliberately waits out the uncertainty: it pauses for the length of the confidence interval before reporting the commit. This ensures that any transaction that starts after the commit will definitely see it.

TrueTime Ordering Simulator

Place events on two nodes. Each event has a confidence interval. The simulation shows whether the events can be ordered or are concurrent.

What If You Don't Have TrueTime?

You're not Google. You don't have atomic clocks. What can you do?

Chapter 5: Process Pauses

You've handled unreliable networks and unreliable clocks. Surely you can at least trust that your own code, running on your own machine, executes continuously? No. Your process can stop running for seconds at any time, without warning, and resume as if nothing happened.

Why Your Code Stops

The Lease Violation Problem

This is where process pauses become dangerous. Consider a lease — a distributed lock with a timeout. A node acquires a lease that says "you are the leader for the next 10 seconds." The node then processes requests, confident that it's the only leader.

Here's what goes wrong:

The Solution: Fencing Tokens

Since you can't prevent a stale leader from acting, you need the storage layer to reject stale writes. The mechanism is a fencing token: a monotonically increasing number issued with each lease grant.

Fencing Token Simulator

This simulation shows two nodes competing for a lease. Node A gets the lease, then suffers a GC pause. Node B acquires the lease during the pause. When Node A wakes up, see what happens with and without fencing tokens.

Python: Fencing Token Check

python
class FencedStorage:
    """Storage server that rejects stale writes using fencing tokens."""

    def __init__(self):
        self.data = {}
        self.max_token = {}  # key -> highest token seen

    def write(self, key, value, fencing_token):
        """Write only if the fencing token is >= the highest seen for this key."""
        if key in self.max_token and fencing_token < self.max_token[key]:
            raise PermissionError(
                f"Stale fencing token: {fencing_token} < {self.max_token[key]}. "
                "Write rejected — you no longer hold the lease."
            )
        self.max_token[key] = fencing_token
        self.data[key] = value
        return True

    def read(self, key):
        return self.data.get(key)

# Demo: the GC pause scenario
storage = FencedStorage()

# Node A gets lease with token 34, writes successfully
storage.write("account_balance", 500, fencing_token=34)
print("Node A wrote 500 with token 34")  # OK

# Node B gets lease with token 35 (after A's lease expired during GC)
storage.write("account_balance", 450, fencing_token=35)
print("Node B wrote 450 with token 35")  # OK

# Node A wakes up from GC, tries to write with old token 34
try:
    storage.write("account_balance", 600, fencing_token=34)
except PermissionError as e:
    print(f"Node A rejected: {e}")  # REJECTED

Chapter 6: The Byzantine Generals Problem

So far, we've assumed nodes are honest but unreliable: they might crash, they might be slow, but they never lie. A crashed node stops sending messages. A slow node eventually responds with the correct answer. What if a node sends the wrong answer? Or sends different answers to different nodes?

This is a Byzantine fault: a node behaves in an arbitrary, potentially malicious way. It might send "yes" to node A and "no" to node B. It might report a different value from what it stored. It might impersonate another node.

The Original Problem

In 1982, Lamport, Shostak, and Pease published a paper that framed this as a military scenario. Several Byzantine generals surround an enemy city. They need to agree on a plan: attack or retreat. They communicate by messenger. Some generals may be traitors who send conflicting messages to sabotage the consensus.

Why 3f + 1?

With 3 generals (A loyal, B loyal, T traitor):

Byzantine Generals Simulator

Four generals try to agree on a plan. One is a traitor. Watch how the loyal generals use majority voting to reach consensus despite the traitor's conflicting messages.

Do You Need Byzantine Fault Tolerance?

Almost certainly not. BFT is expensive — you need 3f+1 nodes to tolerate f faults (versus 2f+1 for crash faults). The extra nodes and message rounds add latency and cost.

Chapter 7: System Models — The Showcase

We've cataloged everything that can go wrong: networks drop messages, clocks drift, processes pause, and nodes can crash (or even lie). Now we need a framework for reasoning about these faults. We need system models.

A system model is a set of assumptions about what kinds of faults can happen. It's the contract between the algorithm designer and reality. "If the world behaves at least this well, then my algorithm guarantees these properties."

Crash Models

How do nodes fail?

Timing Models

How predictable are delays?

The Big Simulation: Distributed System Under Stress

Five nodes in a partially synchronous network running a simplified consensus protocol. You are the chaos monkey. Crash nodes, partition the network, trigger GC pauses, and delay messages. Watch how the system handles each scenario — and when it breaks.

Things to try:

• Crash one follower: The system continues. The leader still has a quorum (3 of 5).
• Crash two followers: The leader loses quorum. Writes stop being accepted.
• Network partition (2-3 split): The side with 3 nodes elects a leader and continues. The side with 2 nodes cannot form a quorum.
• GC pause on leader: The leader misses heartbeats. Followers elect a new leader. When the old leader wakes up, it discovers it's been deposed.
• Delay all messages: Partial synchrony violated. Timeouts fire, elections start, but agreement is delayed (not prevented).

Chapter 8: Knowledge and Truth

Here is a thought experiment that crystallizes the central challenge of distributed systems.

You are a node. You just processed a request and sent a response. But your garbage collector kicked in for 15 seconds right after you sent it. When you wake up, you think only a millisecond has passed. Your lease has expired. Another node is now the leader. But you don't know any of this. From your perspective, nothing happened.

A node cannot trust its own judgment. It might be paused. Its clock might be wrong. Its network might be partitioned. It has no way to tell. This is the fundamental asymmetry of distributed systems: a node can observe its own state, but it cannot observe its own reachability or timeliness from the outside.

The Truth Is Defined by the Quorum

Since no individual node can be trusted to assess reality, the only reliable source of truth is a quorum — a majority of nodes agreeing on a fact.

The FLP Impossibility

In 1985, Fischer, Lynch, and Paterson proved one of the most important results in distributed computing. The FLP impossibility theorem states:

This sounds devastating. If consensus is impossible, how do Raft and Paxos work?

The key word is asynchronous. FLP applies to the fully asynchronous model. Real systems are partially synchronous: they have timeouts, and most of the time the network is well-behaved. Practical consensus algorithms exploit this:

Quorum Decision Simulator

Five nodes try to decide if a particular node is dead. Each has its own view (based on whether it can reach the suspect node). A quorum (3 of 5) must agree. See how partitions can cause different nodes to have different views — but the quorum still makes a consistent decision.

Safety vs. Liveness

Every distributed algorithm makes a choice between two properties:

FLP tells us that in an asynchronous system, you cannot guarantee both safety and liveness in the presence of faults. Practical algorithms choose safety always, liveness when possible. During a partition, the system stops accepting writes (sacrifices liveness) rather than accepting potentially conflicting writes (sacrificing safety).

Chapter 9: Interview Arsenal

This chapter is your cheat sheet. Every failure mode, every mitigation, every pattern you need for distributed systems interviews — organized by dimension.

Failure Mode Reference

System Design Questions

Coding Drills

Drill 1: Idempotency layer

python
import hashlib
import time

class IdempotencyLayer:
    """Prevents duplicate processing of requests."""

    def __init__(self, ttl_seconds=3600):
        self.seen = {}  # idempotency_key -> (result, timestamp)
        self.ttl = ttl_seconds

    def execute(self, idempotency_key, operation):
        """Execute operation only if this key hasn't been seen."""
        self._evict_expired()

        if idempotency_key in self.seen:
            # Already processed — return cached result
            return self.seen[idempotency_key][0]

        # First time seeing this key — execute and cache
        result = operation()
        self.seen[idempotency_key] = (result, time.monotonic())
        return result

    def _evict_expired(self):
        now = time.monotonic()
        expired = [k for k, (_, t) in self.seen.items() if now - t > self.ttl]
        for k in expired:
            del self.seen[k]

# Usage:
idem = IdempotencyLayer()
result = idem.execute("charge-user-123-order-456", lambda: charge_credit_card(50))
# If retried with same key, returns cached result without re-charging

Drill 2: Simple lease with fencing

python
import time
import threading

class LeaseService:
    """Simplified lease service with fencing tokens."""

    def __init__(self):
        self.lock = threading.Lock()
        self.current_holder = None
        self.expiry = 0
        self.next_token = 1

    def acquire(self, node_id, duration_sec=10):
        """Try to acquire lease. Returns (success, fencing_token)."""
        with self.lock:
            now = time.monotonic()
            if self.current_holder is not None and now < self.expiry:
                return False, None  # Lease held by someone else

            token = self.next_token
            self.next_token += 1
            self.current_holder = node_id
            self.expiry = now + duration_sec
            return True, token

    def release(self, node_id):
        with self.lock:
            if self.current_holder == node_id:
                self.current_holder = None
                self.expiry = 0

Debug Scenarios

Key Papers

Chapter 10: Connections

This chapter covered the problems of distributed systems — the hostile environment that algorithms must survive. Now you need the solutions: the algorithms and protocols that provide guarantees despite these faults.

Where This Chapter Fits

The Mental Model

After this chapter, you should have a three-layer mental model:

Summary of Key Principles

Related Lessons

• DDIA Chapter 6: Replication — How to copy data across unreliable nodes
• DDIA Chapter 8: Transactions — Atomic operations in a distributed world