Distributed systems theory

1 · The fallacies of distributed computing

Here's a bug you'll write exactly once, and never forget. Your service calls another service — a quick HTTP request to fetch a user's profile. It works flawlessly for months. Then one afternoon the other service's box gets a little slow, the request never returns, and because you never set a timeout, your thread blocks forever. Threads pile up behind it, your pool exhausts, and a service that had nothing wrong with it falls over — because a call you assumed was instant and reliable was neither. That outage is not a coding mistake in the usual sense. It's a belief mistake: you quietly assumed the network behaves like a function call. It doesn't.

Here's the everyday analogy. Talking to another process on the same machine is like asking your roommate a question — instant, reliable, they're right there. Talking to a service across a network is like mailing a letter to another country: it might arrive late, arrive twice, arrive damaged, or never arrive at all, and you often can't tell which of those happened. The whole discipline of distributed systems is the habit of writing code that stays correct when the letter gets lost — and the fallacies are the catalogue of comforting lies that make you forget the letter can get lost.

The common mistake is treating these as trivia to memorise for an interview. They're not a list — they're a checklist of things that will page you at 3am if you ignore them. "The network is reliable" isn't a philosophical claim; it's the difference between code that retries and code that hangs. Read each fallacy as a question: "what breaks in my design if the opposite is true?" That reframing is the entire value of the list.

Peter Deutsch and colleagues catalogued the false assumptions engineers make when they first build across machines. Every one of them causes a real outage class. The network is not reliable, latency is not zero, bandwidth is not infinite, the network is not secure, topology changes, there is not one administrator, transport cost is not zero, and the network is not homogeneous. The discipline of distributed systems is essentially the art of not believing these.

2 · Consensus — Raft & Paxos

Imagine three friends trying to agree on a restaurant over a group chat where messages randomly arrive late, out of order, or not at all — and one friend's phone keeps dying mid-conversation. How do you ever reach a decision everyone will honour, such that nobody shows up at the wrong place? That awkward little problem, scaled up and made rigorous, is consensus — and it turns out to be one of the deepest problems in computer science, because the messages really can be lost and the participants really can crash at the worst possible moment.

Why does this matter so much? Because the instant you have more than one copy of your data, you have a disagreement waiting to happen. Two nodes both think they're in charge; two writes arrive in different orders on different replicas; a node that was asleep wakes up with a stale view of the world. Consensus is the machinery that lets a group of unreliable machines behave, from the outside, like a single reliable one. The trick that makes it work is deceptively simple: a majority must agree. Because any two majorities of the same cluster must share at least one member (they overlap), a decision blessed by a majority can never be contradicted by a different majority — that overlap is the mathematical backbone of everything in this chapter.

The common mistake is thinking consensus means "everyone agrees." It doesn't — requiring everyone would mean a single dead node freezes the whole system, which is exactly the fragility we're trying to escape. Consensus asks only for a majority, so the cluster keeps making progress as long as more than half its nodes are alive and can talk. Another trap: people hear "Paxos is the hard one, Raft is the easy one" and assume Raft is weaker. It isn't — Raft gives the same safety guarantees; it just packages them as an explicit elected leader and a replicated log, which humans can actually reason about. Keep that framing — majority, not unanimity; understandable, not weaker — as you read the precise version below.

Consensus is getting a set of nodes to agree on a single value (or an ordered log of values) even though some may fail and messages may be lost. It is the foundation of replicated state machines — every node applies the same log in the same order, so they stay identical. Raft is the consensus algorithm designed to be understandable: it elects a single leader that accepts all writes, appends them to a log, and replicates them; an entry is committed once a majority has stored it. Paxos solves the same problem and came first, but its "prepare/promise/accept" phases are notoriously hard to follow — Raft trades nothing in guarantees for a lot in clarity.

3 · Leader election

Think about how a group of people with no chairperson start a meeting. There's an awkward silence — everyone's waiting for someone else to begin. Eventually one impatient person can't stand it any longer, speaks up, and says "okay, I'll run this." If nobody objects and enough people nod along, they're now leading. But if two people pipe up at the exact same instant, they talk over each other, everyone gets confused, and the group has to reset and try again. Raft's leader election is precisely this — with a clever twist that makes the "two people at once" collision rare on purpose.

Why does the cluster need a leader at all? Because funnelling every write through one node is what makes the log simple to keep consistent — there's a single place that decides the order of events, so replicas never have to argue about who came first. The hard part is choosing that leader safely when the old one dies and nobody has a global clock or a referee. The mechanism rests on two ideas: a term (a monotonically increasing number that acts as a logical epoch, so everyone can tell a fresh leader from a stale one) and the majority vote from §2 (a candidate is only leader if more than half the cluster backed it in that term, which guarantees there can never be two legitimate leaders at once).

The genuinely clever bit — and the common point of confusion — is the randomised election timeout. If every follower waited the same fixed time before declaring itself a candidate, they'd all declare simultaneously, split the vote evenly, and nobody would win — a livelock that could repeat forever. So each node instead waits a random interval. Almost always one node's timer fires first, it becomes candidate before the others even wake up, and it wins cleanly. The occasional tie just triggers another round with fresh random timers, which almost certainly breaks it. The mistake to avoid: thinking randomness here is sloppy or non-deterministic hand-waving. It's the opposite — it's a deliberate, load-bearing technique for making a distributed race converge quickly.

Raft keeps time in terms (monotonic election epochs). Each node is a follower, candidate, or leader. If a follower hears nothing from a leader before a randomised election timeout, it becomes a candidate, increments the term, and requests votes. A node grants at most one vote per term; a candidate that collects a majority becomes leader. Randomised timeouts make simultaneous candidacies rare, so a split vote is unlikely and quickly resolved in the next term.

election.py# MODEL: a candidate needs a strict majority of the cluster to become leader.
def majority(n):
    return n // 2 + 1                    # >50% of n nodes

def election(n_nodes, votes_for_candidate):
    need = majority(n_nodes)
    won = votes_for_candidate >= need
    return {"cluster": n_nodes, "need": need,
            "got": votes_for_candidate, "elected": won}

print(election(5, 3))    # 3 of 5 -> elected
print(election(5, 2))    # 2 of 5 -> not elected
print(election(4, 2))    # 2 of 4 is NOT a majority (need 3)

{'cluster': 5, 'need': 3, 'got': 3, 'elected': True}
{'cluster': 5, 'need': 3, 'got': 2, 'elected': False}
{'cluster': 4, 'need': 3, 'got': 2, 'elected': False}

4 · Replication & quorums (R + W > N)

Picture a study group that keeps five identical copies of the class notes — one with each member — so that if someone's laptop dies, the notes survive. Now a subtle question: when Priya updates the notes and then Sam reads them, how do you guarantee Sam sees Priya's edit and not an old copy? If Priya only bothered to sync her change to two friends, and Sam happens to read from the other three, he'll miss it entirely. The fix is beautifully simple once you see it: make the group Priya writes to and the group Sam reads from big enough that they're forced to share at least one person. That shared person always has the latest notes — and that single idea is the whole theory of quorums.

This is where the abstract "majority overlap" from consensus becomes a dial you actually turn. Every replicated store answers two questions per operation: how many replicas must acknowledge a write (call it W) and how many must answer a read (call it R), out of N total copies. If R + W > N, the read set and the write set cannot fit into N nodes without colliding — they must overlap — so a read is guaranteed to catch the newest write. That's strong consistency, bought with one inequality. If R + W ≤ N, there's room for them to miss each other, and a read can return stale data — but each operation talks to fewer nodes, so it's faster and survives more failures.

The mistake almost everyone makes at first is treating "strong consistency" as obviously better and just switching it on. But every replica you require to answer is a replica that can be down and block you. Demand acknowledgements from more nodes and your write gets slower and more fragile the moment any of them hiccups; demand fewer and you trade freshness for speed and resilience. There is no setting that wins on all axes — you're picking a point on a trade-off curve, and the honest engineer picks it per workload (a shopping cart tolerates staleness; a bank balance does not). Systems like Dynamo and Cassandra expose R and W directly precisely so you can make that choice operation by operation. Read the rule below as "how much overlap am I buying, and what availability am I paying for it?"

Data is replicated to N nodes. A write quorum W is how many must acknowledge a write; a read quorum R is how many must respond to a read. The key rule: if R + W > N, the read set and write set must overlap in at least one node, so a read is guaranteed to see the latest write — strong consistency. If R + W ≤ N, a read can miss the newest write — eventual consistency, but with lower latency and higher availability. Tuning R and W is how systems like Dynamo/Cassandra dial the CAP trade-off per operation.

quorum.py# MODEL: does this (N, R, W) configuration guarantee a read sees the latest write?
def quorum_config(N, R, W):
    strong = (R + W > N) and (W > N // 2)   # overlap + single-writer safety
    return {
        "N": N, "R": R, "W": W,
        "read_write_overlap": R + W > N,
        "strongly_consistent": strong,
        "write_tolerates_failures": N - W,   # nodes that can be down for a write
        "read_tolerates_failures": N - R,
    }

import json
print(json.dumps(quorum_config(3, 2, 2)))   # classic strong: R+W=4>3
print(json.dumps(quorum_config(3, 1, 1)))   # fast but eventual: R+W=2<=3

{"N": 3, "R": 2, "W": 2, "read_write_overlap": true, "strongly_consistent": true, "write_tolerates_failures": 1, "read_tolerates_failures": 1}
{"N": 3, "R": 1, "W": 1, "read_write_overlap": false, "strongly_consistent": false, "write_tolerates_failures": 2, "read_tolerates_failures": 2}

5 · Commit protocols — 2PC vs 3PC

Think about how a group of friends commit to a plan when everyone must be in or nobody goes. You don't just announce "we're going!" and hope — you first go around and ask each person "can you make it Saturday?" Only once everyone has said yes do you send the second message: "confirmed, it's on." That two-round dance — ask first, confirm second — is exactly two-phase commit (2PC), the classic way to make several database nodes either all apply a transaction or all abort it, with no node left in a half-done state.

Why can't you just tell everyone "commit" in one shot? Because a node might not be able to — its disk could be full, a constraint could be violated, it might be mid-crash. If you skipped the asking phase and half the nodes committed while half couldn't, your data is now inconsistent across machines, which is the one outcome atomicity exists to prevent. The prepare phase is the group asking "everyone able and willing?" and each participant promising it can commit if told to; the commit phase is the coordinator broadcasting the agreed outcome. It's genuinely correct — when it completes.

Here's the fatal flaw, and it's worth feeling viscerally. Suppose everyone votes yes, and then — right before sending the confirmation — the organiser's phone dies. Every friend is now stuck: they promised to go, they're holding the evening open, but they never heard the final word and they can't reach the organiser. They can't safely commit (maybe someone else backed out and the organiser was about to cancel) and can't safely abort (maybe it's on and they'd be no-shows). This is 2PC's blocking problem: a coordinator crash at the wrong instant leaves participants holding locks indefinitely. Three-phase commit (3PC) adds an extra "pre-commit" round so stuck participants can safely time out and decide for themselves — but it only works if you assume messages and clocks have bounded delays, an assumption the real internet happily violates. The common misconception is that 3PC "fixes" 2PC for production; in practice its assumptions rarely hold, which is why serious systems reach for consensus (§2) instead — a leader can die and a new majority simply elects a replacement and finishes the job. Read the two protocols below with that failure moment in mind: the whole story is what happens when the coordinator dies between the phases.

To make several nodes commit a transaction atomically, two-phase commit (2PC) uses a coordinator: phase 1 prepare (everyone votes yes/no and promises to be able to commit), phase 2 commit/abort (coordinator tells everyone the outcome). 2PC is correct but blocking: if the coordinator crashes after everyone voted yes, the participants are stuck holding locks, unsure whether to commit. Three-phase commit (3PC) inserts a pre-commit phase so participants can safely time out and proceed, making it non-blocking — at the cost of an extra round-trip and assumptions (bounded delays) that rarely hold, which is why real systems prefer consensus (Raft/Paxos) over 3PC.

6 · Logical time — Lamport & vector clocks

You've almost certainly seen this bug in a chat app. Two people reply to a message at nearly the same moment, and the messages show up in a different order on your phone than on your friend's — or a reply appears above the thing it was replying to. The instinct is to fix it with timestamps: just sort by the clock, right? Except every phone's clock is slightly off, and "slightly" across machines is enough to sort events into an order that never actually happened. Wall-clock time is a lie you cannot base correctness on — and that uncomfortable fact is why distributed systems invented a different kind of clock entirely.

The insight is to stop asking "when did this happen?" (an unanswerable question with no shared clock) and instead ask "did this happen because of that?" — a question about causality, which you can answer without synchronised time. If event A sent a message that event B received, then A must have come before B, full stop, regardless of what any clock says. A Lamport clock is the simplest way to capture this: a plain counter that ticks on each event and jumps forward whenever you receive a message, guaranteeing that causes always get smaller numbers than their effects. It's enough to build a consistent total order — but it has a blind spot.

That blind spot is the crucial concept, so meet the analogy for it. Two friends in different cities each text a third friend at the same time, never having talked to each other. Neither text "caused" the other — they're concurrent. A Lamport clock will still hand them different numbers, falsely implying one came first. Vector clocks close the gap: instead of one counter, each node carries a whole vector — one counter per node — so comparing two stamps reveals not just an order but the truth: A happened-before B, B happened-before A, or the two are genuinely concurrent (neither vector dominates the other). The common mistake is assuming "concurrent" is a rare edge case you can ignore; in a replicated store it's the exact signal that two writes conflict and the system must consciously reconcile them rather than silently pick one. Read the mechanics below as a way of detecting that conflict honestly, instead of papering over it with untrustworthy timestamps.

There is no perfectly synchronised wall clock across machines, so you cannot order events by timestamp. Lamport clocks give a counter that respects causality (if A caused B, then A's counter < B's) but cannot tell whether two events are concurrent. Vector clocks fix that: each node keeps a vector of counters (one per node), so you can compare two events and decide happened-before, happened-after, or concurrent — which is exactly what a system needs to detect conflicting writes.

vector_clock.py# MODEL: each node holds a vector of counters. Compare two vectors to decide the
# causal relationship between the events they stamp.
def tick(clock, node):                  # local event on `node`
    c = dict(clock); c[node] = c.get(node, 0) + 1; return c

def merge(local, incoming, node):       # receive a message: merge then tick
    c = {n: max(local.get(n, 0), incoming.get(n, 0)) for n in set(local) | set(incoming)}
    c[node] = c.get(node, 0) + 1; return c

def relation(a, b):
    nodes = set(a) | set(b)
    le = all(a.get(n, 0) <= b.get(n, 0) for n in nodes)
    ge = all(a.get(n, 0) >= b.get(n, 0) for n in nodes)
    if le and ge: return "equal"
    if le:        return "a happened-before b"
    if ge:        return "b happened-before a"
    return "concurrent (conflict!)"

# A does a local event, sends to B; B also did an independent local event
a1 = tick({}, "A")                      # {A:1}
b1 = tick({}, "B")                      # {B:1}  -- independent of a1
b2 = merge(b1, a1, "B")                 # B receives A's message
print("a1 vs b2:", relation(a1, b2))    # a1 happened-before b2
print("a1 vs b1:", relation(a1, b1))    # concurrent -> a real conflict

a1 vs b2: a happened-before b
a1 vs b1: concurrent (conflict!)

7 · Consistency models & CAP → PACELC

Here's a moment you've lived as a user. You change your profile photo, the app cheerfully says "saved," you refresh — and the old photo is back. A minute later it's the new one again. Nothing is broken; you just read from a replica that hadn't caught up yet. That flicker is the whole subject of this section made visible: consistency is the set of promises a distributed store makes about which version of the truth a read is allowed to see. And the surprising, hard-won lesson of the field is that "always show the absolute latest" is not free — it's a premium feature you pay for in latency and availability.

The mental model that unlocks everything is a spectrum, not a switch. At the strong end, linearizable consistency makes the whole cluster behave as if there were a single copy: every read sees the most recent write, in one global order, no flicker ever. At the weak end, eventual consistency promises only that if writes stop, the replicas will eventually agree — in the meantime, reads can be stale (that returning-old-photo moment). Between them sit intermediate promises like causal consistency, which guarantees you'll at least see things in an order that respects cause and effect (your reply never appears before the message it answered), without paying for a full global order. Each step toward "strong" buys you a cleaner mental model and costs you speed and fault-tolerance.

Which brings us to the famous acronym everyone half-remembers. The common mistake is repeating "CAP means pick two of consistency, availability, partition-tolerance" — which is misleading, because partitions aren't optional (the network will split sometimes, per §1), so the real choice is only during a partition: stay consistent or stay available, you can't have both. But CAP quietly hides the everyday cost. PACELC tells the honest, fuller story: Partition → choose A or C; Else (the normal, no-partition case that's true 99.9% of the time) → choose Latency or Consistency. That "Else" clause is exactly the quorum trade-off from §4 generalised: even on a perfectly healthy network, a strongly-consistent read must wait for a quorum, which is slower than reading one nearby replica. Read the models and the CAP→PACELC extension below as the vocabulary for saying precisely what your system promises and what that promise costs on an ordinary Tuesday.

Consistency is a spectrum, not a switch. Linearizable (strongest): every operation appears to happen instantly at a single point — reads always see the latest write. Sequential/causal: weaker orderings that still respect some structure. Eventual (weakest): if writes stop, replicas eventually converge, but a read may be stale in the meantime. Stronger consistency costs latency and availability.

CAP (from SD6): during a network Partition you must choose Consistency or Availability. PACELC extends it honestly: if Partition, choose A or C; Else (normal operation) choose Latency or Consistency. In other words, even with no partition, stronger consistency still costs latency — the quorum trade-off from §4, generalised.

Knowledge check

def majority(n):
    return n // 2 + 1

def wins(n, votes):
    return votes >= majority(n)

print("majority of 5:", majority(5))       # 3
print("3 of 5 wins:", wins(5, 3))          # True
print("2 of 5 wins:", wins(5, 2))          # False
print("2 of 4 wins:", wins(4, 2))          # False (need 3)

def classify(N, R, W):
    overlap = R + W > N
    return {
        "overlap": overlap,
        "consistency": "strong" if overlap else "eventual",
        "write_failures_tolerated": N - W,
        "read_failures_tolerated": N - R,
    }

print("N=3 R=2 W=2:", classify(3, 2, 2))   # strong
print("N=3 R=1 W=1:", classify(3, 1, 1))   # eventual

def tick(clock, node):
    c = dict(clock); c[node] = c.get(node, 0) + 1; return c

def merge(local, incoming, node):
    keys = set(local) | set(incoming)
    c = {n: max(local.get(n, 0), incoming.get(n, 0)) for n in keys}
    c[node] = c.get(node, 0) + 1; return c

def relation(a, b):
    keys = set(a) | set(b)
    le = all(a.get(n, 0) <= b.get(n, 0) for n in keys)
    ge = all(a.get(n, 0) >= b.get(n, 0) for n in keys)
    if le and ge: return "equal"
    if le: return "a before b"
    if ge: return "b before a"
    return "concurrent"

a1 = tick({}, "A")
b1 = tick({}, "B")
b2 = merge(b1, a1, "B")       # B receives A's event
print(relation(a1, b2))       # a before b
print(relation(a1, b1))       # concurrent

import random

def election_round(nodes, term, rng):
    timeouts = {n: rng.uniform(0.15, 0.30) for n in nodes}   # randomised
    candidate = min(timeouts, key=timeouts.get)              # first to time out
    need = len(nodes) // 2 + 1
    # each other node votes for the candidate if it hasn't voted this term
    votes = 1 + sum(1 for n in nodes if n != candidate)      # all grant (no split here)
    if votes >= need:
        return {"term": term, "leader": candidate, "votes": votes}
    return {"term": term, "leader": None, "votes": votes}    # split -> retry

def elect(nodes, seed=1):
    rng = random.Random(seed); term = 0
    while True:
        term += 1
        r = election_round(nodes, term, rng)
        if r["leader"] is not None:
            return r

print(elect(["n1", "n2", "n3", "n4", "n5"]))

def simulate(N, R, W):
    replicas = [{"version": 0, "value": None} for _ in range(N)]
    # W-quorum write of version 1 to the first W replicas
    for i in range(W):
        replicas[i] = {"version": 1, "value": "v1"}
    failed = N - 1                       # take the LAST replica down
    reachable = [i for i in range(N) if i != failed]
    # R-quorum read: consult the LAST R reachable replicas, take highest version
    read_set = reachable[-R:]
    seen = max(replicas[i]["version"] for i in read_set)
    return {
        "N": N, "R": R, "W": W,
        "R+W>N": R + W > N,
        "read_set": read_set,
        "read_saw_version": seen,
        "saw_latest": seen == 1,
    }

print(simulate(3, 2, 2))   # R+W=4>3 -> read sees version 1 even with a node down
print(simulate(3, 1, 1))   # R+W=2<=3 -> read reads replica 1, misses the write

def pacelc(on_partition, on_else):
    """on_partition in {A, C}; on_else in {L, C}. Returns the PACELC label."""
    assert on_partition in ("A", "C") and on_else in ("L", "C")
    return f"P{on_partition}/E{on_else}"

systems = {
    "etcd (Raft, linearizable)":        pacelc("C", "C"),   # PC/EC
    "Cassandra (eventual, tunable)":    pacelc("A", "L"),   # PA/EL
    "Spanner-like (consistent, fast)":  pacelc("C", "L"),   # PC/EL
}
for name, cls in systems.items():
    print(f"{name:34s} -> {cls}")