The Eight Fallacies

1. The network is reliable — packets drop, links flap, partitions happen.
2. Latency is zero — RTT inside region ~1 ms, cross-region 50–200 ms, satellite 600 ms+.
3. Bandwidth is infinite — 10 Gbps NIC = 1.25 GB/s shared by all flows.
4. The network is secure — assume hostile until TLS + authn + authz proven.
5. Topology doesn't change — VMs migrate, DNS shifts, ASGs scale.
6. There is one administrator — multi-team, multi-cloud reality.
7. Transport cost is zero — egress is metered; serialization burns CPU.
8. The network is homogeneous — different vendors, MTUs, jitter, ECMP.

Hierarchy

FLP Impossibility

In an asynchronous system with even one crash failure, no deterministic consensus algorithm can guarantee progress. Real systems sidestep with: timeouts (eventual synchrony), randomization, or partial synchrony assumptions.

Basic Paxos

• Roles: Proposer, Acceptor, Learner. Often colocated.
• Phase 1 (Prepare): Proposer picks ballot n > any seen. Sends Prepare(n) to majority. Acceptor promises not to accept < n; returns highest accepted (n', v').
• Phase 2 (Accept): Proposer sends Accept(n, v) where v = highest v' or own value. Acceptor accepts iff hasn't promised higher.
• Decision: Once majority accept (n, v), value chosen. Learners notified.

Raft

• Strong leader: all writes go through leader; followers replicate.
• Term = monotonically increasing election epoch.
• Log entry = (term, index, command). Committed when replicated to majority.
• Election: follower times out → becomes candidate → requests votes for new term → wins if majority.
• Log matching: if two logs share an entry at index i, all entries before i are identical.
• Safety: leader must contain all committed entries (election restriction).
• Snapshotting: the log grows without bound, so each node periodically snapshots its state machine, discards the prefix, and ships the snapshot to lagging followers via InstallSnapshot when the needed entries have already been compacted away.
• Membership changes: switch configurations through joint consensus (C-old,new), an intermediate config where decisions need majorities in both the old and new sets at once — this prevents two disjoint majorities from electing two leaders during the transition. (Newer Raft variants apply single-server add/remove changes one at a time to get the same overlap guarantee.)

ZAB & Variants

ZooKeeper Atomic Broadcast: total-order broadcast for primary-backup. Used in ZooKeeper. Similar safety to Raft, optimized for read-heavy workloads with watches.

Leases

• Time-bounded grant of authority. Lease holder = leader for duration T.
• If holder fails, others wait T then re-elect. No risk of two writers IF clocks are bounded-skew.
• Renewable: holder refreshes before expiry.
• Use case: Chubby/etcd lock service backing GFS master, Bigtable tablet leader.

Fencing Tokens

// Lock service issues monotonically increasing token on lease grant
client A → lock acquired, token=33
client A → GC pause 30s
client B → lease expired, lock acquired, token=34
client A → resumes, writes "hello" with token=33
storage → rejects: max seen token is 34

Lamport Clocks

• Each process keeps counter L. Increment on local event. On send, attach L. On receive, L = max(L, received) + 1.
• Captures happens-before (→) but not concurrency: a → b ⇒ L(a) < L(b), but inverse not true.
• Use total order via (L, process_id) tiebreak.

Vector Clocks

• Each process i keeps vector V[1..N]. On local event: V[i]++. On send: attach V. On receive: V[j] = max(V[j], received[j]) for all j; then V[i]++.
• Captures concurrency: a || b iff neither V(a) ≤ V(b) nor V(b) ≤ V(a).
• Cost: O(N) per message. Variants: dotted version vectors (Riak), interval tree clocks.

def increment(v, pid):
    v = list(v)
    v[pid] += 1
    return v

def merge(a, b):                       # on receive: element-wise max, then bump self
    return [max(x, y) for x, y in zip(a, b)]

def happens_before(a, b):              # a -> b  iff  a <= b componentwise AND a != b
    return all(x <= y for x, y in zip(a, b)) and any(x < y for x, y in zip(a, b))

def concurrent(a, b):                  # a || b  iff  neither happens-before the other
    return not happens_before(a, b) and not happens_before(b, a)

# P0 does a local event, sends to P1; P2 acts independently -> concurrent with P0's send.
e1 = increment([0, 0, 0], 0)             # P0: [1,0,0]
e2 = increment(merge(e1, [0, 0, 0]), 1)  # P1 receives e1, then bumps: [1,1,0]
e3 = increment([0, 0, 0], 2)             # P2 independent: [0,0,1]
assert happens_before(e1, e2)            # e1 causally precedes e2
assert concurrent(e1, e3)                # e1 and e3 have no causal path

Hybrid Logical Clock (HLC)

Combines physical time with logical counter. Provides one-way causality + tight bounds to wall clock. CockroachDB, YugabyteDB use HLC for MVCC ordering.

Spanner TrueTime

Google's API: TT.now() returns interval [earliest, latest] with bounded uncertainty (typically ~1 ms via GPS+atomic). Spanner waits out uncertainty to provide externally consistent (strict-serializable) transactions. Cost: extra latency = clock uncertainty.

Modes

Quorum Math

R + W > N guarantees at least one replica seen the latest write at read time (strong consistency). R + W ≤ N = eventual. Common: N=3, R=2, W=2.

class Coordinator:
    def __init__(self, n): self.replicas = [(0, None)] * n   # each: (version, value)

    def write(self, ids, version, value):       # ack only after W replicas persist
        assert len(ids) >= self.W
        for i in ids: self.replicas[i] = (version, value)

    def read(self, ids):                         # read R replicas, take newest version
        assert len(ids) >= self.R
        return max((self.replicas[i] for i in ids), key=lambda vv: vv[0])

c = Coordinator(3)                               # N = 3
c.N, c.W, c.R = 3, 2, 2                           # R + W = 4 > N  => quorums overlap

c.write({0, 1}, version=1, value="x")            # write to replicas {0,1}
v, val = c.read({1, 2})                           # read from {1,2}: intersects at replica 1
assert (v, val) == (1, "x")                       # overlap guarantees we see the latest write

Chain Replication

Nodes in a chain: head → middle → tail. Writes start at head, propagate down chain. Reads from tail. Properties: strong consistency, throughput equal to single node, easy failure handling. CRAQ variant allows reads from any node with version check for higher read throughput.

Spectrum (Strongest → Weakest)

CAP / PACELC

CAP: in a partition (P), choose Consistency or Availability. PACELC extends: when no partition (Else), choose Latency or Consistency. Example: Cassandra = AP + EL. Spanner = CP + EC.

Common CRDTs

class GCounter:                         # grow-only counter
    def __init__(self, n): self.p = [0] * n
    def inc(self, i, k=1): self.p[i] += k
    def value(self): return sum(self.p)                 # value = sum of per-replica maxima
    def merge(self, o): self.p = [max(a, b) for a, b in zip(self.p, o.p)]

class PNCounter:                        # supports decrement: two G-Counters
    def __init__(self, n): self.pos, self.neg = GCounter(n), GCounter(n)
    def inc(self, i, k=1): self.pos.inc(i, k)
    def dec(self, i, k=1): self.neg.inc(i, k)
    def value(self): return self.pos.value() - self.neg.value()
    def merge(self, o): self.pos.merge(o.pos); self.neg.merge(o.neg)

a, b = GCounter(3), GCounter(3)         # two replicas of a 3-node counter
a.inc(0); a.inc(0); b.inc(1)           # concurrent increments on different replicas
a.merge(b); b.merge(a)                 # gossip both ways -> converge
assert a.value() == b.value() == 3     # no lost updates

Production Uses

• Riak — OR-Set for shopping carts.
• Redis CRDB — counters across regions.
• Figma / Linear / Notion — text + tree CRDTs for offline-first editing.
• DynamoDB Streams + custom merge — convergent shopping cart.

Two-Phase Commit (2PC)

• Coordinator → Prepare → all participants vote yes/no.
• If all yes: Commit. If any no or timeout: Abort.
• Blocking: if coordinator crashes between phases, participants stuck holding locks.
• 3PC adds pre-commit phase for non-blocking under crash-stop, but assumes synchrony — rarely used in practice.

Sagas

• Long-running business transaction = sequence of local txns + compensating actions.
• Forward path: T1 → T2 → T3. If T3 fails, run C2, C1 to undo.
• Orchestration (central state machine) vs choreography (event-driven). Orchestration easier to debug.
• Not atomic — observers see partial states. Use semantic locks / status fields.

TCC (Try-Confirm-Cancel)

Reservation pattern: Try reserves resources without final commit. Confirm finalizes. Cancel releases. Used in payment + inventory systems.

Transactional Outbox

To publish event + commit DB row atomically: write event to outbox table in same DB txn. Separate process tails outbox, publishes to message bus, marks rows sent. Eliminates dual-write inconsistency.

Delivery Semantics & Idempotency

The outbox relay can crash after publishing but before marking the row sent, so it republishes — delivery is at-least-once, not exactly-once. Consumers must dedup.

Gossip Protocols

• Epidemic spread: each round, node picks k random peers, exchanges state.
• Convergence: O(log N) rounds for full propagation.
• Resilient to churn; no central coordinator.
• Used: Cassandra membership, Consul, Hashicorp Serf, Bitcoin block propagation.

Merkle Trees for Anti-Entropy

• Hash leaves of dataset bottom-up. To compare two replicas, exchange root hash. If equal: identical. If different: descend differing subtrees.
• Bandwidth: O(divergence) instead of O(dataset).
• Used: Dynamo, Cassandra, IPFS, git pack negotiation.

Read Repair & Hinted Handoff

• Read repair: on read, if replicas disagree, write back the latest version to stale replicas.
• Hinted handoff: when target replica is down, coordinator stores hint locally and replays when target recovers.

The Hash Ring: Consistent Hashing

Naive sharding with hash(key) % N breaks catastrophically on scaling: changing N remaps almost every key. Consistent hashing maps both keys and nodes onto the same circular hash space (e.g. 0 .. 2^32-1). A key is owned by the first node found walking clockwise from hash(key). Add or remove one node and only the keys in that node's arc move — the rest stay put.

• Key movement — with N nodes, adding one moves on average only K/N keys (K = total keys), versus ~K·(N-1)/N for modulo hashing. Minimal disruption is the whole point.
• Ownership rule — successor node clockwise. Replication factor R: store on the next R distinct physical nodes walking the ring (the "preference list").
• Lookup — sort node positions once; binary-search the ring for a key's successor in O(log N).

import bisect, hashlib

def h(s: str) -> int:
    return int(hashlib.md5(s.encode()).hexdigest(), 16)

class Ring:
    def __init__(self):
        self.ring = []            # sorted hash positions
        self.owner = {}           # position -> node

    def add(self, node):
        p = h(node); bisect.insort(self.ring, p); self.owner[p] = node

    def locate(self, key):
        p = h(key)
        i = bisect.bisect_right(self.ring, p) % len(self.ring)  # wrap around
        return self.owner[self.ring[i]]

Virtual Nodes & Rebalancing

Plain consistent hashing has two flaws: with few nodes the arcs are wildly uneven (load skew), and when a node dies its entire load dumps onto a single successor (a hot spot). Virtual nodes (vnodes) fix both by placing each physical node at many positions on the ring.

• Vnodes — each physical node gets V tokens (e.g. 128–256). A node's relative load variance shrinks like 1/√V (the coefficient of variation depends on tokens-per-node, not cluster size), so more tokens = flatter distribution.
• Graceful failure — a dead node's V arcs are scattered, so its load spreads across many survivors instead of crushing one neighbor.
• Heterogeneity — a beefier machine simply gets more tokens, absorbing proportionally more keys.

Rendezvous (HRW) Hashing

Rendezvous — or Highest Random Weight (HRW) — hashing solves the same distribution problem without a ring. For a key, compute hash(key, node) for every node and pick the node with the maximum score. To replicate to R nodes, take the top-R scores.

def hrw_locate(key, nodes, r=1):
    scored = sorted(nodes, key=lambda n: h(f"{key}:{n}"), reverse=True)
    return scored[:r]   # top-R = preference list, deterministically ordered

• No ring state — nothing to precompute or keep sorted; any node can independently agree on ownership given the member list.
• Minimal movement — removing a node reassigns only its keys, to their next-highest scorer; no cascade. Same disruption guarantee as consistent hashing.
• Natural replica ordering — the sorted score list is the preference list for free, no ring-walk to dedupe physical nodes.

Byzantine Fault Tolerance & PBFT

Crash-fault tolerance (Raft, Paxos) assumes failed nodes simply stop. Byzantine faults are worse: a node may lie, send conflicting messages to different peers, or be actively adversarial (compromised host, buggy firmware, public blockchain). Tolerating f Byzantine nodes needs 3f+1 total, not 2f+1.

• Why 3f+1 — a quorum is 2f+1. Two quorums of size 2f+1 in a system of 3f+1 intersect in at least f+1 nodes, so at least one honest node is common to both — the intersection guarantees agreement despite f liars. With only 2f+1 total you cannot both tolerate f silent failures and detect f liars.
• PBFT phases — pre-prepare (leader proposes), prepare (replicas cross-check they saw the same proposal), commit (replicas confirm enough peers are prepared). A request commits after 2f+1 matching COMMITs.
• View change — if the primary is faulty, replicas trigger a view change to elect the next primary, carrying proofs of what was already prepared so nothing safe is lost.

PBFT message flow (f=1, N=4):
  Client → Primary            REQUEST
  Primary → all               PRE-PREPARE(seq, digest)
  each replica → all          PREPARE   (wait for 2f matching)
  each replica → all          COMMIT    (wait for 2f+1 matching)
  replicas → Client           REPLY     (client waits for f+1 identical)

Failure Detection: Heartbeat vs Phi-Accrual

Membership systems must decide "is node X dead?" — but the network never tells you; a timeout is a guess. A binary heartbeat detector fixes a threshold: miss it and you declare death. The phi-accrual detector instead outputs a continuous suspicion level, letting each subsystem pick its own sensitivity.

• Heartbeat + fixed timeout — simple, but a static threshold is a lose-lose: too short → false positives on GC pauses/latency spikes; too long → slow detection of real crashes.
• Phi-accrual — track the recent distribution of inter-arrival times; on each check output φ = -log10(P(now - last_arrival)). φ rises smoothly as a heartbeat grows overdue relative to the historical mean/variance.
• Adaptive — a node on a slow link naturally has a wider arrival distribution, so φ tolerates its jitter without extra tuning. Callers act at their own threshold (e.g. φ > 8).

import math
# phi over a normal fit of recent inter-arrival samples
def phi(now, last_arrival, mean, std):
    t = now - last_arrival
    # P(arrival later than t) via Gaussian CDF tail
    p_later = 1.0 - normal_cdf(t, mean, std)
    return -math.log10(max(p_later, 1e-12))
# act when phi > 8  (~10^-8 chance the node is actually alive)

SWIM Membership & Gossip Dissemination

Pushing all-to-all heartbeats is O(N²) and drowns big clusters. SWIM (Scalable Weakly-consistent Infection-style Membership) separates failure detection from membership dissemination, giving constant per-node load regardless of cluster size.

1. Direct probe — each period, a node pings one random member and waits for an ack.
2. Indirect probe — no ack? Ask k other random members to ping the target on your behalf. This survives a single bad link and cuts false positives dramatically.
3. Suspicion — still no ack? Mark the target suspect (not dead) and gossip that. The target, or anyone who can reach it, can refute within a timeout before it is declared dead.
4. Dissemination — membership updates piggyback on the probe/ack messages (infection-style gossip), so a change reaches all N nodes in O(log N) rounds with no dedicated broadcast.

• Constant load — each node sends O(1) messages per period; total is O(N), not O(N²).
• Anti-entropy — periodic full-state gossip repairs anything the infection missed, bounding staleness.
• In the wild — HashiCorp's memberlist (Consul, Nomad) and Uber's Ringpop implement SWIM; incarnation numbers let a suspected node overrule stale rumors of its death.

End-to-End Case Studies

Real systems combine the primitives above. Each row below is a different answer to "how do we stay consistent and available under partitions and node loss."

• Spanner / TrueTime — TrueTime returns an interval [earliest, latest] with bounded uncertainty ε (GPS + atomic clocks). To serialize, a writer waits out the uncertainty: commit_wait until TT.now().earliest > commit_ts, so no later transaction can get a smaller timestamp. Typical ε is a few milliseconds, so writes pay a few-ms commit-wait to buy external (linearizable) consistency globally.
• Kafka ISR — the In-Sync Replica set is followers caught up within replica.lag.time.max.ms. With acks=all a produce is acked only once all ISR members have it. min.insync.replicas=2 means: if ISR shrinks below 2 the leader rejects writes rather than risk data loss — an availability-for-durability trade.
• ZooKeeper / etcd as coordination — not data stores but consensus primitives: leader election, distributed locks, config, and service discovery. They are deliberately small and strongly consistent (majority quorum, linearizable reads) so that other systems (Kafka's controller, Kubernetes' etcd) can offload the hard agreement problem to them.