Distributed systems

CAP is the founding theorem of distributed systems design: when nodes can't talk, you must choose between serving possibly-stale answers or refusing to answer at all. PACELC, introduced by Daniel Abadi in 2012, fills in the missing 99% of life — what happens when there is no partition. Together they give you a 2×2 to place every database you will ever discuss in an interview.

Fischer, Lynch, and Paterson proved in 1985 that you cannot build a deterministic consensus protocol that is both safe (never decides wrong) and live (always decides eventually) in a fully asynchronous network with even a single crash failure. This is the deepest theoretical result in distributed systems. Understanding what it says — and what it doesn't say — is essential for reasoning about why Raft, Paxos, and every real system works the way it does.

Consistency is the most overloaded word in distributed systems: it means one thing in ACID, another in CAP, and varies across a spectrum for replicated databases. This chapter builds two distinct lattices — the replication consistency ladder (what an isolated read can see) and the transaction isolation ladder (what a transaction sees of concurrent transactions) — with concrete anomalies to make each level stick.

-- Doctor A's transaction:
SELECT COUNT(*) FROM on_call WHERE on_duty = TRUE;  -- returns 2 (A and B)
UPDATE on_call SET on_duty = FALSE WHERE doctor = 'A';  -- goes off call

-- Doctor B's transaction (same snapshot):
SELECT COUNT(*) FROM on_call WHERE on_duty = TRUE;  -- also returns 2
UPDATE on_call SET on_duty = FALSE WHERE doctor = 'B';  -- also goes off call

Wall clocks drift. NTP corrects them but introduces ~10ms of uncertainty. At scale, 10ms is an eternity — enough for thousands of operations to interleave incorrectly. Distributed systems need a way to order events without trusting hardware clocks. This chapter covers the four clock strategies you must know: Lamport (total order, no causality), vector clocks (full causality, O(N) cost), HLC (physical anchor + logical tie-break), and TrueTime (bound the uncertainty and sleep through it).

Process A: event a1 → L=1; event a2 → L=2; sends msg to B (carries L=2)
Process B: event b1 → L=1; receives from A → L=max(1,2)+1=3; event b2 → L=4
Process C: event c1 → L=1; event c2 → L=2

Process A: a1 → VC=[1,0,0]; sends to B with [1,0,0]
Process B: b1 → VC=[0,1,0]; receives from A → [max(0,1), max(1,0)+1, 0] = [1,2,0]; b2 → [1,3,0]
Process C: c1 → VC=[0,0,1]; sends to B with [0,0,1]
Process B: receives from C → [max(1,0), 3+1, max(0,1)] = [1,4,1]; b3 → [1,5,1]

2PC — prepare, then commit

1. Coordinator sends PREPARE to all participants.
2. Each participant locks resources, writes a prepare record to its log, votes YES or NO.
3. If all YES, coordinator writes COMMIT to its log, sends COMMIT. If any NO or timeout, sends ABORT.
4. Participants apply and release locks.

The blocking problem: if the coordinator crashes after participants vote YES but before sending COMMIT/ABORT, participants are stuck holding locks indefinitely. They cannot decide unilaterally because they don't know what the coordinator told the others.

3PC and Paxos Commit

3PC adds a pre-commit phase so participants can recover safely if the coordinator dies — but it only works under synchronous network assumptions, which is why it's almost never deployed. Paxos Commit (Gray & Lamport 2006) is the production answer: replicate the coordinator's state machine via Paxos. Spanner does this — each shard is a Paxos group, and 2PC runs across groups.

Sagas — the alternative for long flows

(Garcia-Molina & Salem 1987.) Break a long-running tx into N local transactions, each with a compensating action (refund, cancel, release inventory). On failure mid-flow, run compensations for completed steps in reverse. No distributed lock. Caveat: not isolated — other readers see intermediate states.

Paxos in one paragraph

Two phases. Prepare: a proposer picks a unique number n, asks acceptors to "promise" not to accept anything < n; if it gets a majority of promises, it learns the highest-numbered already-accepted value. Accept: it asks the same majority to accept (n, value); if all promise still holds, the value is chosen. Quorum (majority) ensures any two rounds intersect. Multi-Paxos elides the prepare phase once a leader is stable — the typical case.

Raft — the modern default

Ongaro & Ousterhout, USENIX ATC 2014. Designed top-down for understandability. Splits consensus into three sub-problems: leader election, log replication, safety.

Membership changes: joint consensus — overlap old and new configurations during the transition so any majority of old AND any majority of new must overlap, preserving safety.

Build it yourself

Implement Raft in MIT 6.824 Lab 2 (Go). The single most useful exercise for distributed-systems interviews. After Lab 2 + Lab 3 (KVStore on Raft), you can answer Raft questions from first principles instead of recall.

ZAB — ZooKeeper Atomic Broadcast

Primary-backup with total order. Like Raft, has leader election and log replication, but ZAB explicitly preserves order of all messages from the primary across reconfiguration. Used by ZooKeeper; its semantics (linearizable writes, sequentially-consistent reads, watches) are the basis for Chubby-style coordination services.

Quorum intuition

With N replicas, R read replicas, W write replicas: if R + W > N, every read set and every write set share at least one node (pigeonhole). So a read will see at least one replica that has the latest write. Typical balanced choice: R = W = ⌈N/2⌉ + 1.

Consistent hashing

Hash both keys and nodes onto a ring (e.g., 0..2^32). Each key is owned by the next clockwise node. Adding or removing a node only moves K/N keys, vs nearly all under naive modulo hashing. Virtual nodes (each physical node owns many ring positions) smooth out load when N is small.

Two refinements you should know

• Jump hash (Lamping & Veach 2014) — O(log n) compute, no node IDs needed, no memory state. Limitation: only works for "node count goes from N to N+1" — you can't remove arbitrary nodes mid-ring. Good for stateless services with a known cluster size.
• Rendezvous (HRW) hashing — for each key, hash (key, node_id) for every node; pick the node with the max hash. Same K/N rebalance property as consistent hashing. Trivially supports weighted nodes (multiply hash by weight).

The four topologies

• Single-leader: writes go to the leader, replicate to followers. Sync replication = no data loss but writes block on slowest follower; async = fast but you can lose recently-acked writes on failover. Most production setups use one sync follower (durability) and N async (read scaling).
• Multi-leader: writes can go to any leader; conflicts must be resolved (LWW, CRDTs, or app logic). Useful for multi-datacenter writes when you cannot tolerate cross-region write latency.
• Leaderless (Dynamo-style): any replica accepts writes; gossip + sloppy quorum + hinted handoff for availability; read-repair on conflict; anti-entropy via Merkle tree comparison.
• Chain replication (van Renesse & Schneider 2004): writes flow head → ... → tail; reads from tail. Strong consistency with simple failure recovery (just shorten/extend the chain).

Sync vs async — the durability dial

Sync replication: leader waits for followers to ack before returning OK to the client. No data loss on leader crash, but availability drops if a follower lags. Async: leader returns OK after local write; recently-acked writes can be lost if the leader dies before replication catches up. Semi-sync (MySQL, Postgres): wait for at least one follower to ack — usually the right tradeoff.

The four you should know cold

• G-Counter (grow-only) — array of per-replica counters; increment your own slot; read = sum; merge = element-wise max.
• PN-Counter — two G-Counters (one for increments, one for decrements); read = inc - dec.
• OR-Set (observed-remove) — every add tags the element with a unique ID; remove only erases tags you have actually observed. Concurrent add+remove resolves to "present."
• LWW-Register — timestamped value; latest timestamp wins. Loses concurrent writes — use only when "last write wins" is semantically OK.
• RGA (replicated growable array) — text editing CRDT; each character has a unique ID and reference to its predecessor; insertion is commutative.

CRDT vs OT

Operational Transform (Google Docs) achieves the same goal differently: transform concurrent ops to commute. Requires a server to linearize. CRDTs converge peer-to-peer without one. OT is dominant historically; CRDTs dominate new collaborative apps because they're easier to reason about offline.

The three you should know

• Heartbeats — periodic ping with timeout. Simple, but the timeout choice is brutal: too short and you false-positive; too long and you take forever to detect a real failure.
• Phi-accrual (Hayashibara 2004) — instead of binary, output a continuous suspicion level φ = -log₁₀(probability of mistake). App tunes a threshold. Adapts to observed inter-arrival distribution. Used in Cassandra, Akka.
• SWIM (Das 2002) — Scalable Weakly-consistent Infection-style membership. Each node periodically pings a random peer; on failure, asks K random peers to indirectly probe. Gossips suspicions and confirmations. Scales to thousands of nodes. Used in HashiCorp Serf, Memberlist, Consul.

How LSM works in 4 lines

1. Writes go to a memtable (sorted in-memory) + a write-ahead log on disk.
2. When the memtable fills, it flushes to an SSTable (sorted, immutable file) at level 0.
3. Background compaction merges SSTables across levels (size-tiered or leveled).
4. Reads: check memtable, then each level; bloom filters skip levels that can't contain the key.

Kafka exactly-once — the recipe

(1) Idempotent producer: producer ID + per-partition sequence number → broker dedups retried writes. (2) Transactions: producer wraps writes across partitions; transaction coordinator atomically commits or aborts. Consumers in read_committed isolation only see committed records.

Spanner: why both Paxos AND 2PC

Common confusion: "Spanner has Paxos, why does it need 2PC?" Paxos replicates within a single shard (Paxos group). Cross-shard transactions need 2PC across Paxos groups, where each "participant" is itself a fault-tolerant Paxos group. So 2PC's blocking problem is mitigated — the coordinator and every participant are themselves replicated.

Load balancing

L4 = TCP-level (HAProxy in TCP mode, AWS NLB). L7 = HTTP-level (Envoy, NGINX, AWS ALB) — can route on path/header/cookie, terminate TLS, do retries. Algorithms: round-robin, least-connections, consistent-hash, EWMA.

Power of two choices (P2C) — pick 2 backends at random, send to the less loaded. Mitzenmacher's result: with O(N log log N / log N) max load vs O(log N / log log N) for naive random. Nearly optimal at almost zero coordination cost. Use it.

Caching

• Cache-aside (lazy): app reads cache; on miss, reads DB and populates cache.
• Write-through: writes hit cache and DB synchronously.
• Write-behind: writes hit cache; flushed to DB async. Risk: data loss on cache crash.
• Eviction policies: LRU (most common), LFU, ARC, TinyLFU, S3-FIFO, SIEVE. SIEVE (2024) is the new low-overhead default.

Rate limiting

Three classic shapes:

• Token bucket — steady refill rate, allows bursts up to bucket size. Most flexible.
• Leaky bucket — steady output rate, no burst. Use when downstream can't burst.
• Sliding window — count requests in last T seconds. Approximate variants for memory.

-- KEYS[1] = bucket key; ARGV: rate, capacity, now
local b = redis.call('HMGET', KEYS[1], 'tokens', 'ts')
local tokens, ts = tonumber(b[1]), tonumber(b[2])
local rate, cap, now = tonumber(ARGV[1]), tonumber(ARGV[2]), tonumber(ARGV[3])
if not tokens then tokens, ts = cap, now end
tokens = math.min(cap, tokens + (now - ts) * rate)
local allowed = 0
if tokens >= 1 then tokens = tokens - 1; allowed = 1 end
redis.call('HMSET', KEYS[1], 'tokens', tokens, 'ts', now)
redis.call('EXPIRE', KEYS[1], 3600)
return allowed

Other essentials

• Idempotency keys — client sends a UUID; server stores (key → result); safe retries. Stripe's pattern.
• Circuit breakers / bulkheads / timeouts / jittered exponential retries — Hystrix-style. Avoid retry storms with full jitter (random in [0, backoff]).
• Service mesh — Envoy / Istio sidecar proxies handle mTLS, retries, observability.
• Event sourcing — state = log of events. CQRS = separate read/write models.

ML-specific distributed (covered elsewhere)

Parallel training and inference patterns live on distributed training and LLM inference. Vector DB internals are in ML system design problem 8.

1. Raft (Ongaro & Ousterhout 2014) — read in full
2. Spanner (Corbett 2012)
3. Dynamo (DeCandia 2007)
4. BigTable (Chang 2006)
5. GFS (Ghemawat 2003)
6. MapReduce (Dean 2004)
7. ZooKeeper (Hunt 2010)
8. Kafka (Kreps 2011) — LinkedIn paper
9. Calvin (Thomson 2012) — deterministic distributed DB
10. FLP impossibility (Fischer, Lynch, Paterson 1985)

Distributed systems quiz — readiness check

1. Explain CAP precisely.
   Show answer

   Under network partition, you choose between consistency (linearizability) and availability (every non-failed node responds). P is not optional in real networks — you choose CP or AP during a partition. Common misreading: "pick 2 of 3" — wrong, because real systems must tolerate partitions.

2. Difference between snapshot isolation and serializable?
   Show answer

   SI doesn't prevent write skew: two transactions both reading and updating disjoint rows based on a constraint that depends on the other's reads. SSI (Cahill 2008) adds predicate locks + dependency tracking to detect write skew. PostgreSQL's "serializable" is SSI.

3. How does Spanner achieve external consistency?
   Show answer

   TrueTime gives bounded uncertainty intervals via GPS + atomic clocks. Commit-wait: after assigning a commit timestamp, wait for the uncertainty interval to pass before releasing the commit. Ensures any subsequent transaction sees this commit's timestamp.

4. Walk through Raft.
   Show answer

   Three sub-problems: (1) Leader election: random timeouts → become candidate → request votes → win majority → leader. (2) Log replication: leader appends + replicates + commits when majority ack. (3) Safety: leader's log wins; only commit current-term entries. Membership changes via joint consensus.

5. How does Dynamo handle conflicts?
   Show answer

   Vector clocks identify concurrent versions; client (or LWW) resolves on read. Sloppy quorum + hinted handoff for availability. Read repair: when read sees divergent versions, async write the merged value back. Anti-entropy via Merkle tree comparison between replicas.

6. Compare LSM tree vs B-tree.
   Show answer

   LSM: write-fast (sequential append), read-amp (bloom + multi-level lookup), needs compaction. B-tree: balanced, in-place updates, slower writes. LSM dominates write-heavy KVs (Cassandra, RocksDB). B-tree dominates OLTP (Postgres, InnoDB).

7. Design a rate limiter at 100k req/s per key.
   Show answer

   Token bucket per key in Redis with Lua atomic update: store tokens + last-refill timestamp; on each request, refill based on (now − last) × rate, capped; deduct if enough; return decision. Higher scale: shard Redis or local approximate counters with periodic sync.

8. Kafka exactly-once semantics — how?
   Show answer

   (1) Idempotent producer: PID + per-partition sequence number → broker dedups retries. (2) Transactions: producer wraps writes across partitions in a transaction; transaction coordinator atomically commits/aborts. Consumers in read_committed only see committed messages.

9. Consistent hashing vs jump hash vs rendezvous?
   Show answer

   Consistent hash: ring with virtual nodes; adding a node moves K/N keys. Jump hash (Lamping & Veach 2014): O(log n) compute, no node IDs. Rendezvous (HRW): hash (key, node) for all nodes; pick max — handles weighted nodes naturally.

10. What is FLP and what does it mean for production systems?
    Show answer

    FLP impossibility (Fischer, Lynch, Paterson 1985): in a fully asynchronous network with even one crash, no deterministic protocol guarantees both safety and liveness. Real systems sidestep with timeouts (semi-synchrony assumption), randomization, or accept rare unavailability. Raft and Paxos prefer safety over liveness.

11. Why use vector clocks vs Lamport clocks?
    Show answer

    Lamport clock: total order, doesn't preserve causality fully. Vector clock: per-process counter array; captures full causality (a → b iff VC(a) < VC(b)). Cost: size O(N processes). Use vector when you need to detect concurrent vs causally-ordered events; Lamport when you only need total order.

12. Quorum math: R + W > N — why?
    Show answer

    If R + W > N, every read intersects every write (pigeonhole). So a read sees at least one replica that has the latest write. Strong-ish consistency without coordination. R = W = ⌈N/2⌉+1 is the typical balanced choice.

13. Explain CRDTs and give two examples.
    Show answer

    Conflict-free replicated data types: structures that converge under concurrent updates without coordination. G-Counter: per-replica counter, sum on read. OR-Set: tag adds with unique IDs; deletes only remove tags you've seen. Used in collaborative editors (Figma, Linear) and offline-first apps.

14. Why does Spanner need 2PC despite Paxos?
    Show answer

    Paxos is for replication within a Paxos group (one shard). Cross-shard transactions need 2PC across groups. Spanner: 2PC over Paxos — each "participant" in 2PC is a Paxos group, so the transaction is durable even with single-node failures.

15. What is sloppy quorum + hinted handoff?
    Show answer

    If a designated replica is down, the write goes to the next node in the ring (sloppy — outside the strict quorum). The receiving node holds the data as a "hint" and forwards to the original when it recovers. Tradeoff: improves write availability; risks stale reads on the original.

16. Phi-accrual vs heartbeat for failure detection?
    Show answer

    Heartbeat: ping with timeout; binary alive/dead. Phi-accrual (Hayashibara 2004): probabilistic — outputs a continuous suspicion level φ; threshold tuning per-app. Adapts to network jitter. Used in Cassandra, Akka.

17. Design a distributed lock service.
    Show answer

    Cluster of 5 nodes running Raft. Lock = key in replicated log with TTL + owner. Acquire: CAS (set if not exists); on fail, watch for delete event. Release: delete key. Crash safety: TTL releases abandoned locks. Use fencing tokens (monotonic) to prevent stale clients from acting after lease expiry.

18. Explain MapReduce in one paragraph.
    Show answer

    Distributed computation framework. User writes map(k, v) → list[(k, v)] and reduce(k, list[v]) → output. System partitions input across mappers, shuffles by key, runs reducers on grouped data. Handles fault tolerance (re-run failed tasks), data locality (run map close to data), straggler mitigation (speculative execution).

19. What is read-your-writes consistency?
    Show answer

    A user always reads their own latest writes (even in an eventually-consistent system). Implemented by routing reads to the same replica that handled writes, or by tracking client write timestamps and reading from a replica caught up past that timestamp.

20. What's PACELC and how does it extend CAP?
    Show answer

    PACELC (Abadi 2012): partitioned → choose C or A; else (no partition) → L (latency) or C. Most real systems trade off in normal operation too (not just during partitions). Spanner: CP / EC. Dynamo: AP / EL.