Distributed Systems from first principles
Published
Your bar: redraw the core of this field on a whiteboard from memory — and reason about any async, multi-service system without flinching. Almost everything below is a response to one hard truth, so every level is built to be seen and redrawn, not read. Grounded in Kleppmann’s Designing Data-Intensive Applications 1 and the Cambridge Distributed Systems lectures. 2
One truth holds the whole field. Every technique below is a strategy for living with it:
When a remote call goes silent
you cannot know if it failed, succeeded, or runs on
so you build to be safe despite that
never certain of it
The One Hard Truth — you can never know
Partial failure: some parts work while others fail, and the working parts can’t tell which is which.
The four indistinguishable causes. A request that draws no reply has exactly these four explanations — and from A’s vantage point every one of them looks the same: silence.
Is A set of nodes that interact only by messages over a network, where failure is partial, not total-and-knowable.
Why it’s hard A single machine crashes with a stack trace. A distributed node dies silently while everyone else runs on, unaware.
The trap In causes ③ & ④ the work already happened — the charge went through, the row was written. “Just retry” is loaded.
Two Generals No finite exchange of messages makes two parties certain they agree over a lossy channel. Proven impossible — not hard.
The acknowledgement regress. Each message could be the one that’s lost, so each needs its own confirmation, without end. Certainty is never reached in a finite number of messages. (Two Generals, Kleppmann, Cambridge Lecture 2.)
Timeouts are a guess, not a truth. A timeout doesn’t detect failure; it declares it — and there is no provably-correct value for N:
✓ TCP ACK = “kernel got bytes,” not “app did the work” — the four causes live at the application layer
🧠 Memory rule: When a remote call goes silent you cannot tell lost request · slow node · lost reply · dead node apart — and the work may already be done.
Memory check
- Four indistinguishable causes of no reply? → lost request · node slow/paused · lost reply · node dead
- What is a timeout, really? → a guess that declares failure; it doesn't detect it
- What does Two Generals prove? → certain agreement over a lossy channel is impossible
Delivery guarantees & idempotency
The fix for the double-charge: pick at-least-once, then make processing duplicate-proof.
The three delivery guarantees. Send-and-forget may lose the message; retry-until-acked may duplicate it; exactly-once is the goal you can’t get at the delivery layer.
| Guarantee | Promise | The price · where you see it |
|---|---|---|
| At-most-once | send & forget; 0 or 1 times | may be lost, never duplicated · metrics, telemetry |
| At-least-once | retry until acked | never lost, may duplicate · Kafka, SQS, webhooks (the default) |
Exactly-once delivery | once, period | impossible over an unreliable network — see below |
An unacked send leaves you two moves: resend (risk a duplicate → at-least-once) or don’t (risk a loss → at-most-once). There is no third door — it’s Two Generals in work clothes. So stop chasing exactly-once delivery and build exactly-once effects.
The idempotency key. The timed-out retry carries the same key; the server recognises it, returns the original result, and does not charge again. At-least-once + idempotency = exactly-once effect.
Idempotent Doing it many times = doing it once. balance=100,
DELETE /orders/42, “mark #7 paid.”
Not idempotent balance=balance+10, POST /orders,
“increment counter,” “append a row.”
Idempotency key Client mints one UUID per logical op, reuses it on every
retry; server stores key→result, replays the saved result on repeats.
(Stripe’s Idempotency-Key.)
Dedup (receiver side) Broker tags each message; consumer records processed IDs and drops repeats — exactly-once within the dedup window.
✓ Idempotency defends against duplicates only; reordering needs ordering (Lessons 3–4) or commutative ops
🔁 Memory rule: Exactly-once delivery is a myth; ship at-least-once + idempotency = exactly-once effects.
Memory check
- The default real-world guarantee? → at-least-once (may duplicate)
- Why is exactly-once delivery impossible? → unacked send forces resend-or-not = duplicate-or-loss
- Who generates the idempotency key, and when? → the client, once per logical op, reused on retries
No global clock — logical time
Stop asking when things happened on a wall clock; ask which one caused the other.
| Clock kind | Answers | Failure mode |
|---|---|---|
| Physical / time-of-day | ”what wall-clock instant?“ | skew, drift, backward jumps |
| Monotonic | ”how much elapsed locally?“ | meaningless across machines |
| Logical | “what is the order?“ | carries no real-time meaning |
Happens-before vs concurrent. B sends m2 only after receiving m1, so
send(m1) → recv(m1) → send(m2); C’s event is reachable from neither —
concurrent. (Lamport, 1978.)
Happens-before (a → b) Same node & earlier; or send→receive of one message; plus transitivity. A partial order.
Concurrent (a ∥ b) Neither reaches the other — causally independent, not “same wall-clock time.”
Lamport clock One integer; recv: L=max(L,t)+1. Gives
a→b ⟹ L(a)<L(b) — but not the converse. Can’t detect
concurrency.
Vector clock One counter per node; compare element-wise.
a→b iff V(a)<V(b); incomparable = concurrent. Cost: O(N) per message.
(Dynamo.)
✓ NTP error is tens of ms and can jump backward; use logical time for ordering
⏱ Memory rule: There is no global clock — for ordering, throw away time and track cause: happens-before.
Memory check
- Why are cross-machine timestamps unsafe to order? → skew can make the later event carry the smaller stamp
- What does a Lamport clock fail to do? → tell whether two events were concurrent or causal
- What does a vector clock add? → an exact concurrent-vs-causal test
Order & causality — the broadcast ladder
Pick the cheapest rung that still satisfies the requirement — that choice is the whole art.
The broadcast-guarantee ladder, and the causal violation it prevents. Each rung adds exactly one ordering promise; the causal rung is the one that stops a reply from overtaking the question it depends on.
| Rung | Adds this promise | Concretely |
|---|---|---|
| Best-effort | nothing — try once | sender may crash mid-send; some get it, some don’t |
| Reliable | all-or-nothing | if any correct node delivers, every correct node eventually does |
| FIFO | per-sender order | one sender’s messages arrive in send order |
| Causal | happens-before order | if A→B, every node delivers A before B |
| Total order | one global sequence | all nodes deliver all messages in the same order |
✓ First ask: FIFO, causal, or total? Total order needs consensus; don’t pay for it unless required
📡 Memory rule: FIFO ⊂ causal ⊂ total;
need every replica on one order → you need total-order broadcast → you need consensus
.
Memory check
- Which rung stops a reply overtaking its question? → causal
- Why is total order the expensive jump? → nodes must agree on order of concurrent messages = consensus
- State-machine replication needs what input? → same commands, same order (a total-order log)
Replication — leader, multi-leader, leaderless
Keep copies useful-agreement-close despite the lost messages and timeouts you already accepted.
| Architecture | Who takes writes | Key trade-off |
|---|---|---|
| Single-leader | one leader; followers replay its log | no write conflicts (one serializer); leader is a SPOF → failover & split brain |
| Multi-leader | a leader per DC / per device | local fast writes; conflicts must be resolved after the fact |
| Leaderless (Dynamo) | any replica; client uses quorums | no leader; freshness via |
A read quorum and a write quorum must overlap by ≥1 node. With N=3, W=2, R=2 (R+W=4>3), whichever two the read picks, at least one was in the write set — so the read sees the latest value. W and R are tunable knobs, not fixed laws. (Dynamo, 2007.)
Why replicate Fault tolerance · latency (replica near the user) · throughput (parallel reads). Name which one you’re buying.
Sync vs async Synchronous = durable but one slow follower blocks; async = fast but a leader crash loses acked writes. Most run semi-sync.
Conflict resolution LWW (silently loses data — clocks lie) · app merge (more code, no loss) · CRDTs (deterministic merge, limited shapes).
Replication lag anomalies read-your-writes · monotonic reads · consistent-prefix — each with a standard bounded fix (route to leader / pin replica).
✓ It only guarantees a read sees a recent write; concurrent/partial writes still leak stale values
🗂 Memory rule: One serializer = no conflicts; many writers = resolve later; no leader = R + W > N for overlap.
Memory check
- Why no write conflicts under single-leader? → the leader serializes all writes into one order
- N=5 — which W,R is safe? → W=3,R=3 (6>5); the rest sum to ≤5
- "Comment vanished on reload" is which anomaly? → read-your-own-writes
Consistency models — eventual → causal → linearizable
A contract for what a read may return. Stronger = fewer surprises, more coordination, more latency.
Consistency is a ladder, not a switch. From eventual (weak) up to linearizable (strong). A stronger model permits fewer behaviours — easier to program against, more expensive to provide. (Jepsen, “Consistency Models”.)
| Model | Promise | Cost / note |
|---|---|---|
| Eventual | if writes stop, replicas converge | any past/stale value meanwhile; cheapest, stays available (AP) |
| Session (RYW, monotonic reads) | per-your-client promises | kills the anomalies users notice; pin to an up-to-date replica |
| Causal | cause-and-effect seen in order everywhere | strongest model that stays available under partition; concurrent writes unordered |
| Linearizable | acts like one copy; recency, real-time order | no staleness window — but needs coordination (consensus/quorum) → latency |
“Consistency” is overloaded CAP’s C (recency of single-object reads) ≠ ACID’s C (transaction invariants) ≠ serializability (multi-object). Keep them apart.
The rule of thumb Buy the strongest model your correctness actually needs and not a rung more. Reserve linearizability for locks, leader election, unique constraints.
✓ Linearizable is recency for single objects; serializable is isolation across transactions — different words
📈 Memory rule: Climb only as high as correctness demands — eventual → session → causal → linearizable, each rung costs coordination.
Memory check
- Strongest model that stays available under partition? → causal
- Cheap per-client fix for "comment vanished"? → session guarantee: read-your-writes
- Why does linearizability cost latency even with no partition? → a node must coordinate to be sure it isn't stale
CAP & PACELC — the real trade-offs
CAP isn’t “pick 2 of 3.” P is forced on you; the only choice is C-vs-A during a partition.
The PACELC decision tree. A partition forces the CAP choice (PA vs PC); but the far more common no-partition state still forces a Latency-vs-Consistency choice (EL vs EC). Most systems are PA/EL or PC/EC. (Abadi, IEEE Computer 2012.)
C / A / P C = linearizability · A = every non-failed node answers (not “fast”) · P = partition tolerance — not optional, partitions happen.
Under partition CP refuses/blocks on the minority side (etcd, ZooKeeper, Mongo default). AP answers with possibly-stale data (Cassandra, DynamoDB, Riak).
PACELC if-Partition: A vs C. Else: L vs C — strong consistency costs round-trips to a quorum even on a healthy network.
Tunable per-request Cassandra/Dynamo let one cluster be EL for feeds and EC for
balances (consistency level / ConsistentRead).
✓ P is mandatory → only choose C-or-A under partition; CAP is silent on healthy networks — that’s the hole PACELC fills (Brewer 2012)
⚖️ Memory rule: if Partition → A or C; Else → Latency or Consistency. Coordination buys consistency and always costs latency.
Memory check
- When does CAP force the C-vs-A choice? → only during a network partition
- What does a CP system do under partition? → sacrifice availability to keep data consistent
- What does PACELC add over CAP? → the latency-vs-consistency trade on healthy networks
Consensus — FLP impossibility + Raft
Get nodes to agree on one value despite crashes — using a majority quorum that always overlaps.
A 5-node Raft cluster electing a leader: one node’s randomized election timeout fires first, it becomes a candidate in a new term, gathers a majority of votes, and becomes leader. Random timeouts dissolve split votes. (Ongaro & Ousterhout, 2014.)
A leader replicating a log entry: it is committed only after a majority persists it, then applied to each node’s state machine. The leader replies to the client only after commit — so an acked write can never be lost.
Consensus = 3 properties Agreement (no two decide differently) · Validity (decided value was proposed) · Termination (everyone eventually decides).
Majority quorum ⌈(N+1)/2⌉. Two majorities always share a node → two conflicting decisions impossible. N=3 tolerates 1, N=5 tolerates 2.
Why odd-sized A 4-node cluster needs 3 for majority, tolerates 1 — same as 3 nodes. The extra even node buys nothing.
Log Matching AppendEntries carries the prior entry’s index+term; a mismatch is rejected and the divergent tail overwritten — committed history never lost.
✓ They assume crash-stop, not lying; malicious nodes need Byzantine consensus (PBFT, 3f+1) (Castro & Liskov 1999)
🗳 Memory rule: elect by randomized timeout · commit on a majority · majorities overlap → two leaders / two decisions can’t coexist.
Memory check
- Three properties of consensus? → agreement, validity, termination
- What does FLP prove? → no deterministic protocol guarantees consensus under async + 1 crash
- When is a Raft entry committed? → once a majority persists it; then it's applied to the state machine
Partitioning / sharding — splitting without hot spots
Replication raises availability; partitioning raises the storage and write ceiling.
Two ways to map keys to partitions, and what each costs. Key-range keeps keys sorted (cheap range scans, but a write burst to one range overloads a node); hash scatters keys evenly (balanced load, but a range scan must hit every node).
| Property | Key-range | Hash |
|---|---|---|
| Range scan | cheap — few adjacent partitions | expensive — scatter to all |
| Load distribution | uneven → hot spots | even by design |
| Example systems | HBase, Bigtable | Cassandra, Dynamo |
Rebalancing — never mod N hash(key) mod N moves almost every key
when N changes. Use a fixed partition count, or consistent hashing (relocates
~K/N).
Hot spot / skew One celebrity key melts one partition; hashing can’t help (same key hashes the same). Fix: random suffix splits the hot key across 100 sub-keys.
Request routing Forward via any node (gossip) · a routing tier · partition-aware client. The map is kept consistent via ZooKeeper/etcd.
Rebalance deliberately Auto-rebalance triggered by a node that merely looks dead (Lesson 1!) can start a data-movement storm.
✓ Every replica holds the whole dataset; to raise the storage ceiling you must partition
🧩 Memory rule: range = cheap scans, hot spots · hash = even load, no scans; grow with fixed partitions or consistent hashing, never mod N.
Memory check
- Failure mode of keying by timestamp + key-range? → all current writes hit one partition (write hot spot)
- Why is
mod Nbad on growth? → changing N reassigns almost every key at once - How to relieve a celebrity's hot partition? → add a random suffix to split the hot key
Putting it together — the resilience toolkit
Every pattern is one trade: at-least-once delivery + idempotent effects = “effectively-once.”
The outbox pattern turns a dual write into one atomic write + an idempotent publish. The service commits the business row and an outbox row in one DB transaction; a relay publishes unsent rows at-least-once and marks them sent.
Dual-write problem Writing DB and publishing as two steps: a crash in the gap leaves them disagreeing. The outbox collapses it into one atomic write.
Effectively-once There’s no exactly-once delivery. Ship at-least-once delivery + idempotent effects. Always ask: exactly-once across which boundary?
End-to-end argument A correctness guarantee (dedup, idempotency) can only be enforced by the endpoints; the network is a best-effort accelerator. (Saltzer, Reed & Clark, 1984.)
Retry budgets Independent retries multiply down a chain (3×3×3 = 27 calls). Cap retries (~10% of volume) and retry at one layer. (Google SRE.)
✓ Only your consumer knows it already processed event abc-123 — put dedup at the endpoint
retry idempotent ops with backoff+jitter · outbox the dual write · saga + compensate · enforce correctness at the endpoints.
Memory check
- Why add jitter to backoff? → spreads synchronized retries so herds stop colliding
- What does the outbox solve? → the dual-write problem — one atomic write + idempotent publish
- Where must dedup live, and why? → at the endpoint; only it knows what it already processed
Interview — pick your bar
Answer out loud in ~60s, then reveal. Core = recall · Senior = trade-offs & failure modes · Staff = synthesis under ambiguity · System Design = open design round (a different axis, not a harder level).
State CAP precisely, then correct the most common misreading.
Name the consistency models from weakest to strongest and what each guarantees.
What is a quorum, and why does R + W > N matter in a leaderless store?
Name the three replication topologies and the price each pays.
A payment call times out with no reply. What do you do, and why is "just retry" dangerous?
Why is "exactly-once delivery" impossible, and what do we actually ship instead?
Two replicas accepted writes; which one wins? Walk me through the clock issue.
You key a time-series table by timestamp and one node is melting. Why, and how do you fix it?
mod N — adding a node remaps almost everything; use a fixed number of partitions or consistent hashing so adding a node moves only ~K/N keys → for read-heavy time ranges, consider a compound key (e.g. tenant + time bucket) so writes spread but scans stay local.Events are "coming out of order." Diagnose it, then decide how much ordering to actually buy.
How does Raft avoid two leaders, and what does FLP have to do with it?
How would you choose a failure detector, and why is "perfect failure detection" the wrong goal?
- Clarify requirements: consistency model needed, ordering, durability, RPO/RTO, scale & latency budget.
- Place it on CAP/PACELC — what happens to each operation under a partition?
- Replication & partitioning: leader topology, quorum (R/W vs N), hash vs range, rebalancing.
- Failure handling: timeouts + backoff + jitter + retry budget; idempotency keys on every side effect.
- Cross-service atomicity: transactional outbox for the dual-write; sagas with compensations (no 2PC across services).
- Observability & ops: detect partitions, lag, hot spots; fencing tokens; how you test rare interleavings.
- End-to-end argument: place each correctness guarantee at the layer that has the knowledge to enforce it.
Design an order-checkout flow across payment, inventory, and shipping that survives partial failure.
Design a globally replicated key-value store and state its consistency contract under partition.
Retrieval quiz — answer before peeking
Effortful recall beats re-reading. Pick one; the right answer highlights with a one-line why.
Q1. Your call times out with no reply. How many fundamentally indistinguishable causes are there?
Q2. "Exactly-once delivery" over an unreliable network is best understood as…
Q3. A leaderless store has N = 5. Which W, R guarantees a read overlaps every recent write?
Q4. CAP forces a choice between Consistency and Availability…
Q5. In Raft, a log entry is <em>committed</em> when…
Q6. A celebrity account melts one partition even under hash partitioning. The standard fix is to…
Reconstruct the field from a blank page
Write the ten lessons in order, one line each, with nothing on screen. Then reveal and compare — gaps are your next drill.
1 Can’t tell slow from dead (4 causes; Two Generals) · 2 at-least-once + idempotency = exactly-once effect · 3 no global clock → happens-before / vector clocks · 4 broadcast ladder; total order ≈ consensus · 5 leader / multi-leader / leaderless; R+W>N · 6 eventual→session→causal→linearizable · 7 CAP (C-or-A under partition) + PACELC (L-or-C else) · 8 FLP impossibility + Raft (majority overlap) · 9 range vs hash; consistent hashing; hot spots · 10 retries+jitter · outbox · saga · end-to-end argument.
I’m your teacher — ask me anything. Say “grill me on distributed systems” for mixed no-clue questions, drop a real system and I’ll place it on the CAP/PACELC and consistency spectrums with you, or ask for a worked space-time diagram. Want the next lesson — spaced, interleaved drills? Just ask.
★ Cheat sheet — buzzword → engineering
Mental models: can’t tell slow from dead · timeout = a guess · exactly-once = at-least-once + idempotency · no global clock → cause not time · FIFO ⊂ causal ⊂ total · R+W>N overlaps · climb consistency only as far as needed · P is mandatory (C-or-A under partition) · majorities overlap · range vs hash · outbox kills the dual write · enforce correctness at the endpoints.
Translation table
| Buzzword | What it actually is |
|---|---|
| Eventual consistency | if writes stop, replicas converge — meanwhile any stale read is legal |
| Quorum | R + W > N so read and write sets share a node |
| CP / AP | under partition: block the minority / serve stale data |
| PACELC | if Partition: A-vs-C; Else: Latency-vs-Consistency |
| Consensus | agree on one value despite crashes; ≈ total-order broadcast ≈ a replicated log |
| Raft commit | entry persisted on a majority, then applied to the state machine |
| Consistent hashing | nodes & keys on a ring; add a node, move only ~K/N keys |
| Exactly-once | marketing for at-least-once delivery + idempotent effects |
| Outbox / saga | one atomic local write + relay / local txns + compensations |
Three instincts to install
① At every network boundary ask “what if this times out but succeeded?” ② Buy the weakest guarantee (delivery, ordering, consistency) your correctness actually needs. ③ Put each correctness guarantee at the layer that has the knowledge to enforce it (the end-to-end argument).
★ Review guides & what to read next
5-min (weekly) Don’t re-read — recall. Say the one hard truth; name the four causes; redraw the broadcast ladder and the R+W>N overlap; recite CAP-vs-PACELC; state when a Raft entry commits.
30-min (when stalled) Blank-page reconstruct all ten lessons; re-derive why each builds on Lesson 1; then place one real database (CP/AP? PACELC? which consistency model? partitioning scheme?).
Read next, in order:①
Kleppmann — Cambridge Distributed Systems notes
(Lessons 1–4) · ② Designing Data-Intensive Applications , Ch. 5, 8, 9 (replication, trouble, consensus) · ③ The Raft visualization & paper (Lesson 8) · ④ Jepsen consistency models map (Lesson 6). Course site → MIT 6.5840.
📖 Primary source: Martin Kleppmann, Designing Data-Intensive Applications + the Cambridge lecture series — the highest-trust grounding for this track.
These notes reflect my current understanding and are updated as I learn, build, and discover better explanations.