Distributed systems theory

1 · The fallacies of distributed computing

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

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

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)

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

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

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

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}")