Track C — Warmup: Every Distributed Primitive, From Zero
Self-contained. Read this and nothing else, and you should be able to walk into a design round, name the two hardest components of any prompt, explain the mechanism of every primitive you propose, and defend the alternative you rejected.
The reported anti-pattern for this round is naming technologies without defending the tradeoff. Every section below therefore ends with what it costs and when it loses, because that is the half candidates skip.
Table of Contents
- Chapter 0: What a Design Round Measures
- Chapter 1: The Arithmetic You Do Out Loud
- Chapter 2: Failure, and What "Handled" Means
- Chapter 3: Time, and Why You Cannot Trust It
- Chapter 4: Leases, Fencing, and the Zombie
- Chapter 5: Replication and Quorums
- Chapter 6: Consensus — Raft at Usable Depth
- Chapter 7: Consistency Models
- Chapter 8: Partitioning and Placement
- Chapter 9: Delivery Semantics and the Outbox
- Chapter 10: Control Under Load
- Chapter 11: CRDTs, When You Can Avoid Coordination
- The Twelve Design Prompts
- The Forty Questions
- References
Chapter 0: What a Design Round Measures
0.1 The five things being scored
In descending order of weight. Internalize the order, because it tells you where to spend your 45 minutes.
1. Did you identify the right hard parts? Every design prompt has two components where it could genuinely fail, and a dozen where it could not. A beautiful deep dive on the wrong component scores worse than a rough one on the right component. This is judgement, and it is the thing they are actually buying.
2. Do your failure sections have all three legs? Detection, containment, recovery. "It retries" is not a failure analysis — it does not say how you know it failed, or what stops the failure spreading.
3. Did you do arithmetic? Any number beats no numbers. "50k/min is ~830/s, so at 20/s per worker that's 42 workers plus headroom, call it 60" is worth more than a paragraph of adjectives, because it demonstrates you can size a system rather than describe one.
4. Did you reject something explicitly? The reported anti-pattern is name-dropping technologies without defending the tradeoff. The antidote is a section that says "I considered X, rejected it because Y, and here's what would flip my choice."
5. Did you stay at the right altitude? A class hierarchy for the job model is too low. "We'll use a queue" with no mention of visibility timeouts is too high. The right altitude is components, their contracts, and where they break.
Notably absent: whether your design matches the interviewer's. It does not have to.
0.2 The template, and the clock
1. Requirements and scope — functional, non-functional, explicitly out of scope
2. Scale numbers — what I assumed and the arithmetic I did
3. API surface — the 3-5 calls that matter
4. Data model — keys and indexes, and WHY those keys
5. High-level architecture — the diagram
6. Deep dive: the two hardest — not the easy ones
7. Failure and recovery — detection, containment, recovery, for each
8. Bottlenecks and evolution — what breaks first at 10x
9. Tradeoffs I explicitly rejected
The 45-minute budget:
| Minutes | Activity |
|---|---|
| 0–5 | Clarify. Requirements, scope, scale numbers — written down |
| 5–10 | API and data model |
| 10–20 | Architecture and the diagram |
| 20–35 | Deep dive on the two hardest components |
| 35–45 | Failure modes, bottlenecks, rejected alternatives |
If you are still drawing boxes at minute 25 the round is lost, regardless of how good the boxes are. Drawing feels like progress, which is exactly why it eats the clock. Decide what is hard before you draw.
0.3 How to find the two hardest components
A repeatable procedure, not intuition. Ask four questions of the prompt:
1. Where does state have to be agreed on by more than one machine? That is where consensus, leases, or conflict resolution live, and it is almost always one of the two hard parts.
2. Where can the system lose data? Every boundary where a message moves between two systems that can fail independently. The answer is durability, idempotency, or an outbox.
3. What is the highest-cardinality or highest-rate thing? That is what determines partitioning, and whether the design survives 10×.
4. Where does one tenant's behaviour affect another's? Isolation and fairness. Usually the deep dive nobody does, and always a strong one.
Worked examples:
| Prompt | Hard part 1 | Hard part 2 |
|---|---|---|
| Job scheduler | Exactly-once dispatch under scheduler failure | Worker liveness, leases, split brain |
| Webhook delivery | Per-destination isolation and backpressure | At-least-once + idempotency + DLQ replay |
| Rate limiter | Atomic distributed check-and-decrement | Fail-open vs fail-closed under store outage |
| URL shortener | ID generation without coordination | Read path caching / hot key |
| News feed | Fan-out on write vs read, and the celebrity problem | Ranking freshness vs cost |
| Design ChatGPT | GPU scheduling and admission | Autoscaling on non-stationary token load |
Say your choice out loud at minute 10: "I think the two places this can actually fail are X and Y, so that's where I want to spend the time — does that match what you care about?" That sentence is worth several points on its own, and it lets the interviewer redirect you before you have burned fifteen minutes.
Chapter 1: The Arithmetic You Do Out Loud
1.1 Little's law
The single most useful equation in system design:
\[ L = \lambda W \]
- L — average number of items in the system (concurrency, queue depth)
- λ — average arrival rate
- W — average time an item spends in the system
It requires almost nothing: a stable system where arrivals equal departures over the long run. No assumption about distributions, service order, or anything else. That generality is why it applies everywhere.
Three ways to use it:
Sizing concurrency. 50,000 requests/sec at 8 ms each → L = 50000 × 0.008 = 400 requests in
flight. If a worker handles 200 concurrently, that is 2 workers minimum.
Deriving latency from queue depth. A queue of 10,000 items draining at 100/s → W = L/λ = 100 seconds. Every newly-enqueued item waits 100 seconds. This is why an unbounded queue is
not a buffer — it is a latency amplifier.
Sizing a connection pool. 500 queries/sec at 20 ms → L = 10 connections busy on average.
Pool of 10 is at 100% utilization with zero headroom; see the next section for why that is a
disaster.
1.2 The utilization knee
For an M/M/1 queue — Poisson arrivals, exponential service, one server — the mean response time is:
\[ W = \frac{W_s}{1 - \rho} \]
where \( W_s \) is service time and \( \rho \) is utilization.
| ρ | Response time | Increase from previous row |
|---|---|---|
| 0.50 | 2.0 × service | — |
| 0.70 | 3.3 × | +65% |
| 0.80 | 5.0 × | +52% |
| 0.90 | 10.0 × | +100% |
| 0.95 | 20.0 × | +100% |
| 0.99 | 100.0 × | +400% |
Latency is hyperbolic in utilization, not linear. Going from 50% to 80% costs 2.5×. Going from 90% to 95% costs another 2×. That is why a service at 85% looks fine on a dashboard and falls over at 92% — and it is the entire quantitative argument for admission control.
Two caveats to state if pressed, because they make the argument stronger, not weaker:
- Real traffic is burstier than Poisson, so the true knee arrives earlier than this table says.
- With c servers (M/M/c) the knee is softer — pooling helps — which is a real argument for fewer, larger pools rather than many small ones. That is the queueing-theory justification for shared thread pools, and it is the counter-argument to bulkheads that you should acknowledge.
1.3 Numbers to have memorized
| Operation | Order of magnitude |
|---|---|
| L1 cache reference | 1 ns |
| Branch mispredict | 3 ns |
| L2 cache reference | 4 ns |
| Mutex lock/unlock | 17 ns |
| Main memory reference | 100 ns |
| Compress 1 KB (snappy) | 2 µs |
| Read 1 MB sequentially from memory | 3 µs |
| SSD random read | 16–100 µs |
| Read 1 MB sequentially from SSD | 49 µs |
| Round trip within a datacenter | 500 µs |
| Read 1 MB sequentially from disk | 825 µs |
| Disk seek (spinning) | 10 ms |
| Round trip US cross-country | 40–70 ms |
| Round trip US ↔ Europe | 80–150 ms |
Two derived rules worth more than the table:
- Memory is ~100× faster than SSD; SSD is ~100× faster than a disk seek.
- Any cross-service hop costs ≥ 0.5 ms, so five sequential hops have a 2.5 ms floor before doing any work. That is the arithmetic behind "fan out, don't chain", and it is why deep synchronous call graphs are a latency bug by construction.
Storage rules of thumb: 1 million × 1 KB = 1 GB. 1 billion × 1 KB = 1 TB. A day is ~86,400 seconds; call it 10⁵. A month is ~2.5 × 10⁶ seconds.
1.4 A worked sizing, start to finish
"Design a service handling 50,000 requests/sec, 4 KB responses, 10 billion stored rows of 512 bytes." Do this out loud in about 90 seconds.
Compute. 50k/s × 8 ms = 400 concurrent. At 200 per box, 2 boxes. But I will not run at 100%: at 60% target that is ~3.3, and to survive losing one of three AZs I need 1.5×, so 6 boxes. If it is CPU-bound at 8 ms of CPU: 50,000 × 0.008 = 400 cores busy, /0.6 = 667 cores, on 16-core boxes = 42 boxes. The difference between those two answers is the entire question of whether the work is I/O or CPU, which is why I would ask.
Storage. 10¹⁰ × 512 B = 5.1 TB logical. × 3 replicas = 15.4 TB. × 1.5 for index, WAL and compaction overhead = ~23 TB. At 8 TB usable per node, 3 nodes for capacity — but probably more for throughput, and I would size on whichever binds.
Network. 50k/s × 4 KB = 200 MB/s = 1.6 Gbit/s. That is 16% of a 10 Gbit NIC — fine. But egress at $0.09/GB is 200 MB/s × 2.6 M s/month = 520 TB/month ≈ $47k/month, which is probably the largest line item in the design and worth saying out loud.
Where it breaks first. At 10× the compute scales linearly, but the storage write path does not — 500k writes/sec against a partitioned store means the hot partition becomes the limit, so partition key choice is the thing to get right now rather than later.
That whole paragraph takes ninety seconds and it is worth more than fifteen minutes of architecture description.
Chapter 2: Failure, and What "Handled" Means
2.1 The three legs
Every failure in your design needs three answers. Missing any one means it is not handled.
Detection — how do you know it happened? A failure you cannot detect is a failure you cannot respond to, and the most dangerous failures are the quiet ones.
Containment — what stops it spreading? A failure that takes down its neighbours is not one failure, it is an outage.
Recovery — how does the system get back to healthy, and does it do so automatically?
"It retries" answers none of these. It does not say how you knew to retry, what stops the retry from amplifying the problem, or what happens when retries are exhausted.
2.2 Fail-stop versus fail-slow
The distinction that separates designs that work in production from designs that work on a whiteboard.
Fail-stop: the node stops. It stops answering health checks, connections are refused, everything is obvious. Easy to detect, easy to handle.
Fail-slow (also gray failure, limping hardware): the node keeps answering — but slowly, or wrongly. A disk with rising latency due to bad sectors. A NIC dropping 5% of packets. A GC death-spiral. A node whose clock has drifted.
Fail-slow is worse, and it is the common case. The node still answers your health check in 2 ms, so your load balancer keeps sending it traffic while real requests take 40 seconds. Every client that lands on it stalls, connection pools fill with waiting requests, and the failure propagates to healthy nodes through their saturated pools.
How to actually detect it:
- Health check the real work path, not a
/healthendpoint that returns 200 unconditionally. - Eject on latency percentiles, not on liveness. If p99 on this node is 10× the fleet median, take it out regardless of what it says about itself.
- Use outlier detection — compare each node against its peers rather than against a fixed threshold, because the threshold is wrong at 3am and wrong again during a traffic spike.
- Bound everything with timeouts, because a timeout converts an unbounded fail-slow into a bounded fail-stop, which you already know how to handle.
If a design's only health signal is "does it respond", it has not handled the common case, and saying so unprompted is a strong signal.
2.3 The failure catalog
| Failure | Detection | Containment | Recovery |
|---|---|---|---|
| Node crash | Heartbeat / lease expiry | Traffic drains to healthy nodes | Replacement joins; state re-replicates |
| Node fail-slow | Latency percentiles vs peers; outlier detection | Eject on latency SLO breach; timeouts everywhere | Restart; investigate. Never trust its self-report |
| Network partition | Quorum loss on the minority side | Minority refuses writes | Merge on heal; reconcile |
| Zombie lease holder | Undetectable — see Ch.4 | Fencing token rejected at the storage layer | Nothing to recover, if fencing worked |
| Thundering herd after outage | Queue depth spike; connection surge | Rate-limited catch-up; jittered restarts | Drain at a bounded rate |
| Retry storm | Request rate rising while success rate falls | Retry budget as a fraction of base traffic | Circuit break, then half-open probe |
| Poison message | Attempt count exceeded | Dead-letter after N attempts | Manual or automated redrive |
| Hot partition | Per-key/per-partition metrics | Split, cache, or rate-limit that key | Rebalance |
| Cascading failure | Correlated latency across services | Bulkheads; timeouts; shedding | Shed until stable, then ramp |
| Data corruption | Checksums; invariant audits | Quarantine; stop replicating it | Restore from a known-good point |
| Clock skew | Skew monitoring vs NTP | Treat a skewed node as unhealthy | Resync; re-elect if it was leader |
| Config rollout gone bad | Canary metrics diverging | Staged rollout; automatic rollback | Revert; the config store is a SPOF too |
| Dependency outage | Error rate; circuit state | Fallback, cached, or degraded response | Half-open probe |
2.4 Deliberately accepting a failure mode
The move that reads as staff rather than senior.
Claiming to have handled everything is falsifiable in one question. Instead, name a failure mode you are choosing not to handle, and say why:
"If we lose an entire region mid-write, in-flight requests to that region are lost. I'm accepting that: cross-region synchronous replication would add 80 ms to every write, which blows the 200 ms p99 budget for the 99.99% of the time there is no regional failure. Instead I replicate asynchronously and expose an RPO of about 5 seconds. If the business needs RPO zero, that is a different design and a different latency budget."
That paragraph demonstrates four things at once: you know the failure exists, you know what handling it would cost, you made a decision with a number attached, and you know what would change it. Almost nobody does this, and it is one sentence.
Chapter 3: Time, and Why You Cannot Trust It
3.1 Wall clock versus monotonic
Two different clocks with two different jobs, and conflating them is a real bug.
Wall clock (time.time(), CLOCK_REALTIME) — "what time is it?" Can jump backwards or
forwards when NTP corrects it, when a VM is live-migrated, or across a leap second. Comparable
across machines, approximately.
Monotonic clock (time.monotonic(), CLOCK_MONOTONIC) — "how much time has elapsed?" Never
goes backwards. Its zero point is arbitrary and it is meaningless across machines.
The rule: measure durations with the monotonic clock; express timestamps with the wall clock.
The bug this prevents: a lease expiry computed as wall_now > lease_expiry_wall. NTP steps the
clock forward by 3 seconds; every lease expires simultaneously; every worker believes it lost
its lock; chaos. Compute lease expiry as elapsed monotonic time on the node that owns the
decision, and the NTP step is irrelevant.
3.2 Clock skew and what NTP guarantees
NTP typically holds machines within a few milliseconds to tens of milliseconds on a good LAN, and much worse across the internet or under load. That is a statistical claim, not a bound. There is no guarantee, and there is no way for a node to know its own skew.
Consequences you must respect:
- Never compare timestamps generated on different machines to decide ordering. Two events 5 ms apart may be recorded in the wrong order.
- Never use "latest wall-clock timestamp wins" to resolve conflicts without acknowledging that a skewed node can silently win every conflict forever, deleting other nodes' writes. This is last-write-wins, and it is a documented data-loss mode in Cassandra deployments with clock problems.
- Monitor skew and treat an out-of-bound node as unhealthy. If it is the leader, force an election.
3.3 Logical clocks and happens-before
If physical time is untrustworthy, use logical time — Lamport, 1978.
Define happens-before (→):
- If a and b are in the same process and a comes first, then a → b.
- If a is a send and b is the matching receive, then a → b.
- Transitive: a → b and b → c implies a → c.
If neither a → b nor b → a, the events are concurrent — and that is a real relationship, not an unknown one. Concurrency here means "no causal path connects them", so no ordering between them is more correct than any other.
Lamport timestamps implement it with one counter per node:
on local event: counter += 1
on send: counter += 1; attach counter
on receive(ts): counter = max(counter, ts) + 1
Guarantee: a → b implies L(a) < L(b).
The limitation, which is the exam question: the converse does not hold. L(a) < L(b) does
not imply a → b; they might be concurrent. So Lamport timestamps give you a total order
consistent with causality, which is enough for tie-breaking and for building a consistent
global order — but they cannot detect concurrency, and therefore cannot tell you when a
conflict needs resolving.
3.4 Vector clocks
To detect concurrency you need a vector: one counter per node.
on local event at node i: V[i] += 1
on send from i: V[i] += 1; attach the whole V
on receive at j of W: V = elementwise-max(V, W); V[j] += 1
Compare two vectors:
V < W(V happened before W) iff every elementV[k] ≤ W[k]and at least one is strictly less.- Neither
V < WnorW < V→ concurrent, and you have a genuine conflict.
That is the power: vector clocks detect conflicts, so the system can surface them (Dynamo's sibling versions) or resolve them deterministically (a CRDT merge).
The cost, which is why they are not everywhere: the vector is O(nodes), it must be stored with every value and sent with every message, and pruning entries for departed nodes is genuinely tricky — prune wrong and you lose the ability to detect a conflict.
3.5 Hybrid logical clocks
HLC (Kulkarni et al., 2014) combines both: a physical component that tracks wall clock closely, plus a logical counter for tie-breaking.
on local/send event:
l' = max(l, physical_now)
if l' == l: c += 1 else: c = 0; l = l'
on receive (l_m, c_m):
l' = max(l, l_m, physical_now)
if l' == l == l_m: c = max(c, c_m) + 1
elif l' == l: c = c + 1
elif l' == l_m: c = c_m + 1
else: c = 0
l = l'
What you get:
- Timestamps are close to physical time (bounded by clock skew), so they are human-readable and usable for "give me everything since 10:00".
- They respect causality: if a → b then HLC(a) < HLC(b).
- Constant size — two integers — unlike a vector clock.
The cost is that they still cannot detect concurrency the way vector clocks can. They give a causally-consistent total order, not conflict detection.
This is what CockroachDB and MongoDB use, and it is the right default answer for "how do you order events across nodes without an atomic clock."
3.6 TrueTime and commit-wait
Spanner's approach: build the uncertainty into the API. TT.now() returns an interval
[earliest, latest] guaranteed to contain the true time, made narrow (a few milliseconds) by
GPS receivers and atomic clocks in every datacenter.
Then commit-wait: after picking a commit timestamp s, deliberately wait until
TT.now().earliest > s before releasing locks. That guarantees that any transaction starting
afterwards gets a strictly later timestamp, which makes timestamps globally meaningful and
gives external consistency (linearizability) over the whole database.
The insight worth taking away: Spanner does not eliminate clock uncertainty, it bounds it and then pays for it in latency — every commit waits out the uncertainty window. Without special hardware that window is tens or hundreds of milliseconds, which is why everyone else uses HLCs and accepts a weaker guarantee.
Chapter 4: Leases, Fencing, and the Zombie
This chapter is the highest-value 1,000 words in the track. Fencing is the thing candidates almost never mention, and it is the difference between a design that is safe and one that silently corrupts data.
4.1 What a lease is and why it expires
A lock grants exclusive access until released. If the holder dies without releasing it, the lock is held forever and the system is stuck.
A lease is a lock with a timeout. "You have this for 30 seconds." If the holder dies, the lease expires and someone else takes it. Liveness restored.
Leases are everywhere: leader election, job claims, queue visibility timeouts, distributed locks, DHCP.
The lease duration is a real tradeoff:
- Short lease (1 s): fast failover, but a GC pause or a network blip causes spurious expiry, so you get lease churn and thrash.
- Long lease (60 s): stable, but a dead holder blocks its work for a full minute.
Which is why holders renew — heartbeat at, say, one third of the lease duration, so two missed heartbeats are tolerable before expiry.
4.2 The zombie problem, in full
Here is the scenario, and you should be able to narrate it from memory:
t=0 Worker A acquires a 30-second lease on job J. Starts work.
t=10 Worker A enters a stop-the-world GC pause. (Or: its NIC drops.
Or: the hypervisor deschedules it. Or: it swaps.)
t=30 The lease expires. A is unreachable and has not renewed.
t=31 The scheduler grants the lease to Worker B. B starts job J.
t=45 B finishes and writes the result.
t=50 Worker A wakes up. From A's perspective, NOTHING HAPPENED.
It has no idea 40 seconds passed. It finishes job J and writes.
Now A's stale write lands after B's correct one, and it silently overwrites it.
The critical point, and the one to say out loud: you cannot detect this. From the scheduler's side, an unreachable worker and a dead worker are indistinguishable — that is a theorem, not an implementation gap. From A's side, it has no way to know it was paused; the pause is invisible from inside.
"A should check whether its lease is still valid before writing" does not work either. Between the check and the write there is a window, and the pause can land in that window. The problem is not the check, it is that the check and the write are not atomic with respect to time.
4.3 Fencing tokens
The fix, and it is beautiful because it needs no detection at all.
Every lease grant carries a monotonically increasing token:
t=0 A acquires the lease. Token = 33.
t=31 B acquires the lease. Token = 34.
t=45 B writes with token 34. Storage records "highest seen = 34". Accepted.
t=50 A writes with token 33. Storage sees 33 < 34. REJECTED.
The zombie's write is refused, not because anyone detected the zombie, but because the token proves it is stale. The storage layer enforces the invariant "never accept a write with a token lower than the highest you have seen", and correctness follows without any liveness assumption whatsoever.
This is why the token must be monotonic: it is the only thing carrying the ordering
information. It is naturally produced by anything that already sequences operations — a Raft
log index, a ZooKeeper zxid, a database sequence.
4.4 Where the token must be checked
This is where most answers go wrong, and it is a great follow-up to volunteer.
The token must be checked by the resource being protected — the storage layer, the file system, the downstream API. Not by the lock service, and not by the client.
If the client checks its own token, you have gained nothing: the zombie client believes its token is current, because from its perspective no time passed.
So fencing is not a property you can add by adopting a lock service. It requires the resource to participate:
UPDATE results
SET value = %s, fence = %s
WHERE job_id = %s AND fence < %s;
-- 0 rows updated means a newer holder already wrote. Do not retry.
If the resource cannot participate — a third-party API with no conditional write, an append-only S3 bucket without preconditions — then you cannot fence, and you must say so and choose a different mitigation: make the operation idempotent so a duplicate is harmless, or accept at-most-once and the possibility of a dropped job.
Redlock is worth knowing about here, because it is a plausible-sounding answer and the critique is instructive. It attempts distributed locking across N independent Redis nodes with a majority quorum. Martin Kleppmann's critique is that it relies on bounded clock drift and bounded pauses for correctness, and neither is guaranteed — a GC pause still produces the zombie above. Antirez's rebuttal is that with fencing tokens, or for efficiency rather than correctness, it is fine. The lesson for you: the safety argument lives in fencing, not in the lock protocol.
Chapter 5: Replication and Quorums
5.1 Three replication modes
| Mode | Ack when | On leader failure | Latency |
|---|---|---|---|
| Synchronous | all replicas have it | zero data loss | slowest replica sets your latency |
| Asynchronous | leader has it | recently acked writes can be lost | fastest |
| Semi-synchronous | ≥ k replicas have it | lose only if > k fail together | one slow replica tolerated |
The honest framing: this is a direct trade between write latency and RPO (recovery point objective — how much data you accept losing).
Asynchronous replication means a client can receive "committed", the leader can die one millisecond later, and that write is gone — with the client believing it succeeded. That is not a bug; it is the contract. It must be stated, and if it is unacceptable the answer is semi-sync and a slower write path.
Semi-sync (wait for one of two followers) is usually the right default: it survives any single node failure with no data loss, and it does not let one slow replica set your p99.
5.2 Quorums, derived
With N replicas, require W acknowledgements to write and R to read.
If W + R > N, the read set and the write set must overlap in at least one replica — pigeonhole. That replica has the latest write, so a read that takes the highest version among its responses sees it.
| N | W | R | Property |
|---|---|---|---|
| 3 | 2 | 2 | Standard. Tolerates 1 failure for both reads and writes |
| 3 | 3 | 1 | Fast reads, no write availability if any node is down |
| 3 | 1 | 3 | Fast writes, no read availability if any node is down |
| 5 | 3 | 3 | Tolerates 2 failures |
| 3 | 1 | 1 | W + R = 2 ≤ 3 — no overlap, eventual consistency only |
W = R = ⌈(N+1)/2⌉ is the balanced choice, tolerating ⌊(N−1)/2⌋ failures.
5.3 What a quorum does not give you
The follow-up that separates people who have read about quorums from people who have used them. Kleppmann's DDIA Chapter 5 lists these; know at least three:
- Sloppy quorums break the guarantee. If, under partition, writes are accepted by any W reachable nodes rather than the W "home" nodes, the overlap argument no longer holds. Dynamo does this deliberately for availability, and it means quorum reads may miss recent writes.
- Concurrent writes still need conflict resolution. Two writes with no causal relationship both satisfy W; the quorum does not order them. You need last-write-wins (lossy) or version vectors (correct, more work).
- A write that fails after reaching some replicas is not rolled back. If W=2, one replica accepts and the second fails, the client gets an error — but the first replica keeps the value, and a later read may return it. The client thinks it failed; the system disagrees.
- Read-your-writes is not guaranteed across sessions unless you pin the client to a replica or track the version it last wrote.
- Quorum reads are not linearizable in general without a read-repair-then-commit step, because two concurrent readers can see different values.
Saying "quorums give you overlap, not linearizability" is the compressed version.
5.4 Read repair and anti-entropy
Two mechanisms for converging replicas that have diverged:
Read repair — when a read finds replicas disagreeing, write the newest value back to the stale ones. Cheap, and it happens on the read path, so it repairs exactly what is being used. Its weakness is that rarely-read data is never repaired.
Anti-entropy — a background process compares replicas and reconciles. Comparing everything is expensive, so use a Merkle tree: a hash tree over the key range. Two replicas compare root hashes; if equal, they are identical and the comparison cost was one hash. If not, descend into the differing subtree. Cost is O(log n) in the size of the difference rather than O(n) in the size of the data.
The pairing is the point: read repair for hot data, anti-entropy for cold. Neither alone is sufficient, and Dynamo-lineage systems run both.
Chapter 6: Consensus — Raft at Usable Depth
6.1 What problem consensus actually solves
Consensus lets a group of nodes agree on one value, even when some fail. Built up, it lets them agree on an ordered sequence of values — a replicated log. And a replicated log is enough to build anything: apply the same operations in the same order to the same initial state and every replica reaches the same state. That is the replicated state machine approach, and it is why consensus is the foundation under etcd, ZooKeeper, Consul, and the metadata layer of most distributed databases.
FLP impossibility (1985): in a fully asynchronous system with even one faulty process, no deterministic algorithm guarantees consensus. Practical systems escape this by adding partial synchrony — timeouts — which buys liveness probabilistically while keeping safety unconditionally. Raft never returns a wrong answer; under sufficiently bad conditions it may fail to return one.
That distinction — safety always, liveness under assumptions — is the correct one-sentence summary of every practical consensus protocol.
6.2 Terms, elections, and split votes
Raft divides time into terms, each with at most one leader. A term is a logical clock: it increases monotonically, and every message carries the sender's term. If a node sees a higher term, it steps down to follower immediately. That single rule prevents most split-brain cases by construction.
Three states: follower, candidate, leader.
Election. A follower that hears nothing from a leader for its election timeout increments its term, becomes a candidate, votes for itself, and requests votes. A node grants its vote if (a) it has not voted in this term, and (b) the candidate's log is at least as up to date as its own. A candidate with a majority becomes leader.
Split votes. Two candidates could each get half. Raft handles it with randomized election timeouts (typically 150–300 ms): each node picks a random value, so one usually times out first and wins before the others start. If a split does occur, the term ends with no leader and a new election begins — with fresh random timeouts. This is a probabilistic solution to a symmetry problem, and it is elegant precisely because it needs no coordination.
6.3 Log replication and the commit index
The leader takes client requests, appends to its log, and replicates via AppendEntries (which
doubles as the heartbeat).
An entry is committed once it is on a majority of nodes. The leader tracks a
commitIndex and piggybacks it on subsequent messages so followers learn what is safe to apply.
Log matching: each AppendEntries includes the index and term of the entry preceding the
new ones. A follower rejects it unless it has a matching entry there. From this the protocol
derives an invariant by induction: if two logs contain an entry with the same index and term,
the logs are identical up to that point.
That invariant is why recovery is simple. A new leader does not need to reason about arbitrary divergence; it walks backwards until it finds the last matching entry and overwrites everything after it.
6.4 The two safety rules that make it work
Both exist to prevent a committed entry from ever being lost.
Rule 1 — The election restriction. A voter refuses its vote to a candidate whose log is less up to date than its own ("up to date" = higher last term, or equal term and longer log). Since a committed entry is on a majority, and a winning candidate needs a majority, the two majorities must intersect. So any node that can win an election already has every committed entry.
Rule 2 — Never commit an entry from a previous term by counting replicas. This one is subtle and it is the classic Raft interview question. A new leader may see an entry from an older term replicated on a majority. It is tempting to declare it committed. It is not safe — there are interleavings where such an entry can still be overwritten. Raft's rule: a leader only commits entries from its own term by counting; older entries become committed indirectly, when a newer entry from the current term commits above them. In practice a leader appends a no-op at the start of its term specifically to trigger this.
Being able to explain Rule 2 is a genuine signal, because it is the part people skip.
6.5 What consensus costs
Say these numbers, because "we'll use Raft" without them is exactly the name-dropping anti-pattern:
- Every write is at least one round trip to a majority. Same-DC that is ~1 ms; cross-region it is 50–150 ms. Consensus across regions makes writes slow, always.
- The leader is a throughput ceiling. All writes go through one node. Scale by sharding into many Raft groups — which is what CockroachDB, TiKV and Spanner all do — not by making one group faster.
- Failover is a latency spike, not a seamless handover. Election timeout plus election time is typically hundreds of milliseconds of unavailability for writes.
- You need an odd number — 3 or 5. Going 3 → 4 does not improve fault tolerance (both tolerate 1 failure) and makes every write wait for more nodes.
- Membership changes are the hard part. Naive reconfiguration can produce two disjoint majorities. Raft uses joint consensus, or single-node-at-a-time changes.
Therefore: use consensus for metadata, not for data. Which shard owns which range, who the leader is, cluster membership, configuration. Push the data path onto simpler replication. Saying this shows you know what consensus is for.
6.6 Paxos, briefly and honestly
Single-decree Paxos decides one value with prepare/promise then accept/accepted phases, and a majority at each. Multi-Paxos amortizes the first phase across a stable leader — at which point it is structurally very similar to Raft.
The honest comparison: Paxos was first and is what most pre-2014 systems (Chubby, Spanner) build on; Raft was designed for understandability and specifies leader election, membership changes, and log compaction concretely, which is why almost every post-2014 implementation is Raft.
Do not pretend to deeper Paxos knowledge than you have. "I know Raft properly and Paxos at the level of what problem it solves and why Raft replaced it in practice" is a good, credible answer. Fumbling a half-remembered ballot-number argument is not.
Chapter 7: Consistency Models
7.1 Linearizability
Definition: the system behaves as if there is a single copy of the data and every operation takes effect atomically at some instant between its invocation and its response.
The consequence people actually care about: once a write completes, every subsequent read — from anyone — sees it or something newer. No stale reads, ever.
It is a recency guarantee about single objects. It is what you want for a lock service, a leader election, a unique-ID allocator, or a counter that must not go backwards.
The cost: every read must confirm it is not stale, which means either reading through the leader, or a quorum read plus repair, or a lease. That is a network round trip on the read path, and it is unavailable during a partition on the minority side.
7.2 Serializability
Definition: the result of executing transactions concurrently is equivalent to some serial order of those transactions.
It is an isolation guarantee about multi-object transactions. It says nothing about which serial order, and in particular nothing about real time. A serializable system may legally order a transaction that started at 10:00 after one that started at 10:05, as long as the outcome matches some serial schedule.
7.3 Why they are different words
This is a favourite question and the answer is crisp:
| Linearizability | Serializability | |
|---|---|---|
| About | single objects | multi-object transactions |
| Guarantees | recency — real-time ordering | isolation — equivalent to some serial order |
| Says nothing about | transactions | real time |
Strict serializability is both: equivalent to some serial order, and that order respects real time. That is what Spanner provides and what people usually mean when they say "strongly consistent". It is the most expensive guarantee and the reason Spanner needs TrueTime.
7.4 The weaker models you will actually ship
| Model | Guarantee | Typical use |
|---|---|---|
| Eventual consistency | replicas converge if writes stop | DNS, Dynamo-style stores |
| Read-your-writes | you see your own writes | user profile after an edit |
| Monotonic reads | you never go backwards in time | a feed that must not un-load posts |
| Consistent prefix | you see writes in causal order, maybe not all | chat: never a reply before its message |
| Causal consistency | causally related ops are ordered everywhere | collaborative editing |
| Bounded staleness | at most T behind | dashboards, analytics |
Session guarantees (read-your-writes, monotonic reads) are the ones users actually notice, and they are much cheaper than linearizability — usually implemented by pinning a session to a replica, or by having the client carry the version it last observed and the replica waiting until it has caught up to it.
Proposing session consistency where linearizability is not required, and saying why, is a strong answer. Reaching for "strong consistency" reflexively is not.
7.5 CAP, stated correctly
CAP is widely misquoted. The precise statement:
When a network partition occurs, a system must choose between consistency (linearizability) and availability (every non-failing node answers).
The clarifications that matter:
- It only applies during a partition. With no partition you can have both. "CP or AP" as a permanent property of a system is a category error.
- "Available" in CAP means every non-failing node responds — a very strong definition. A system that stays up but returns errors from the minority is "unavailable" in CAP terms while being perfectly fine in operational terms.
- PACELC is the more useful framing (Abadi): if Partition, then A or C; Else, then Latency or Consistency. It captures the everyday tradeoff — even with no partition, stronger consistency costs latency — which is the tradeoff you actually make every day.
Say PACELC. It signals you have read past the blog-post version.
Chapter 8: Partitioning and Placement
8.1 Hash versus range
Hash partitioning — partition = hash(key) % N.
- Even distribution, near-automatically.
- Range queries are impossible — adjacent keys land on unrelated partitions.
- Adding a node with plain modulo remaps almost everything, which is why you need consistent hashing.
Range partitioning — partition by key ranges: A–F, G–M, N–Z.
- Range queries are efficient — they touch few partitions.
- Hot spots are easy to create: timestamp-prefixed keys send every write to the newest partition, which is the classic HBase/Bigtable mistake.
- Needs dynamic splitting as ranges grow.
Choose by query shape, not by taste. Range queries required → range partitioning, plus a plan for hot spots (salt the prefix, or split aggressively). No range queries → hash.
8.2 Consistent hashing, derived
The problem with hash(key) % N: change N from 4 to 5 and roughly 80% of keys move. For a
cache that is a total flush; for a database it is a full reshuffle.
Consistent hashing fixes it. Map both keys and nodes onto the same circular hash space (0 to 2³²−1). A key belongs to the first node encountered walking clockwise from the key's position.
Now add a node. It lands somewhere on the ring and takes over only the arc between itself and its counter-clockwise predecessor. Only K/N keys move, and only from one neighbour. Remove a node, and only its arc moves, to its clockwise successor.
That is the whole idea: N changes cost O(K/N) movement instead of O(K).
8.3 Virtual nodes
Plain consistent hashing has two problems:
- Uneven load. Random placement of a few nodes gives arcs of very different sizes — some nodes get several times their share.
- Uneven failure impact. When a node dies, its entire arc moves to one successor, which may then be overloaded and fall over too. That is a cascade.
Virtual nodes fix both: each physical node gets many positions on the ring (100–256 is typical).
- Load evens out by averaging over many arcs — variance drops as 1/√v.
- When a node dies, its many small arcs are redistributed across many successors, so no single node absorbs the whole load.
- Heterogeneous hardware becomes easy: give a bigger machine more virtual nodes.
Saying "consistent hashing with virtual nodes, because otherwise a node failure dumps its whole range on one successor and cascades" is a complete answer.
8.4 Hot partitions
Even distribution of keys does not give even distribution of load. One celebrity account, one viral product, one huge tenant.
Options, in order of preference:
- Cache it. The hottest key is by definition the most cacheable. Often the entire answer.
- Split the key.
celebrity_id→celebrity_id:0..99, writes to a random shard, reads fan out and merge. Trades read cost for write distribution. - Dedicated partition. Give the hot key its own resources; treat it as a special case.
- Rate-limit it. Sometimes correct: one tenant's traffic should not degrade everyone.
The requirement that comes first: per-key metrics. You cannot fix a hot partition you cannot see, and aggregate metrics hide it completely — a partition at 100% while the fleet averages 30%. Mentioning that detection precedes mitigation is a good instinct to show.
8.5 Rebalancing without downtime
Moving a partition while serving traffic:
- Snapshot the source partition and copy it to the destination.
- Stream the delta — writes that arrived during the copy.
- When the delta is small, briefly block writes to the range, drain, and flip ownership.
- Route new requests to the destination; the source keeps serving reads until routing propagates.
The hard part is step 3, and it must be fenced. During the flip both nodes believe they own the range for some window. A fencing token in the routing epoch means the old owner's writes are rejected after the flip. Without it you get lost writes at exactly the moment you are trying to be careful.
Two design rules worth stating:
- Never rebalance automatically on a node failure. A node that is briefly unreachable triggers a massive rebalance, which loads the cluster, which makes more nodes unreachable. Rebalance on operator action or after a long, deliberate delay.
- Rate-limit the rebalance. Copying at full speed competes with production traffic. Cap it and accept that rebalancing takes hours.
Chapter 9: Delivery Semantics and the Outbox
9.1 The three semantics
At-most-once — send, do not retry. Messages can be lost, never duplicated. At-least-once — retry until acked. Messages can be duplicated, never lost. Exactly-once — cannot be achieved for delivery.
The impossibility: sender sends, receiver processes, ack is lost. The sender cannot distinguish "never arrived" from "arrived, ack lost". Resend risks a duplicate; do not resend risks a loss. Adding round trips just moves the problem to the ack of the ack. This is the Two Generals problem.
What is achievable is exactly-once processing: at-least-once delivery plus an idempotent consumer. Say it that way. Systems that advertise "exactly-once" (Kafka's transactions, Flink's checkpointing) are doing exactly this — at-least-once plus deduplication inside a transactional boundary they control — and they are careful to scope the claim to their own boundary.
The requirement that follows: a stable idempotency key, generated by the producer,
unchanged across retries. Generate it at send time and every retry has a new key, so dedupe
silently does nothing. This is precisely why Stripe's API makes the client supply
Idempotency-Key.
9.2 The dual-write problem
The most common distributed bug, and it hides in code that looks obviously correct:
db.save(order) # 1
queue.publish(OrderCreated(order)) # 2
Crash between 1 and 2: the order exists and nobody downstream knows. Swap the order and a crash leaves an event for an order that does not exist. Wrap them in a transaction and it does not help — the queue is not in your database's transaction.
There is no ordering of two independent writes that is safe. That is the point.
9.3 The outbox pattern
Make it one write.
BEGIN;
INSERT INTO orders (...) VALUES (...);
INSERT INTO outbox (id, topic, payload, created_at)
VALUES (gen_random_uuid(), 'orders', %s, now());
COMMIT;
Both rows are in the same database transaction, so they commit or roll back together. A separate relay process then reads the outbox and publishes:
SELECT id, topic, payload FROM outbox
WHERE published_at IS NULL
ORDER BY created_at
FOR UPDATE SKIP LOCKED
LIMIT 100;
-- publish, then mark published_at = now()
Properties:
- Atomic — the event exists iff the state change exists.
- At-least-once — the relay may crash after publishing and before marking, so a message can repeat. Consumers must be idempotent. That is fine; it is the semantics you can actually have.
- Ordered per aggregate if you order by the aggregate's sequence.
The alternative is change data capture: tail the database's replication log (Debezium reading the Postgres WAL or MySQL binlog) and publish from there. No outbox table and no relay polling, at the cost of coupling to the database's replication format and needing a CDC pipeline. Both are correct; the outbox is simpler to reason about and to test.
FOR UPDATE SKIP LOCKED is worth calling out: it lets many relay workers claim disjoint rows
without blocking each other, and it is the same primitive that makes Postgres a perfectly good
job queue at moderate scale.
9.4 Dead-letter queues done properly
After N failed attempts, a message goes to a dead-letter queue so it stops consuming capacity and blocking the main flow.
A DLQ is only useful with all four of these, and most designs mention only the first:
- The reason. Store the last error, the stack, and the attempt count with the message. A DLQ full of payloads and no diagnosis is a landfill.
- A replay path. An operator must be able to fix the cause and re-inject. Design it now, not during the incident.
- Poison-message detection. A message that fails deterministically — malformed payload, deleted referent — should go to the DLQ fast rather than burning N retries with backoff. Distinguish permanent failures (4xx-shaped) from transient ones (5xx, timeout) and do not retry the permanent ones at all.
- Alerting on the rate. A DLQ nobody watches is a silent data-loss channel. Alert on arrival rate, not depth, because depth only tells you about the past.
And name the ordering consequence: if you need per-key ordering and message 5 dead-letters, message 6 either blocks (head-of-line blocking, preserving order) or proceeds (breaking order). There is no third option. Choose deliberately and say which.
Chapter 10: Control Under Load
10.1 Retry storms and budgets
Retries are the most common way a partial outage becomes a total one.
With k attempts and failure probability f, offered load is multiplied by:
\[ \sum_{i=0}^{k-1} f^i \]
| Failure rate | 3 attempts | 5 attempts |
|---|---|---|
| 10% | 1.11× | 1.11× |
| 50% | 1.75× | 1.94× |
| 80% | 2.44× | 3.36× |
| 95% | 2.85× | 4.52× |
| 100% | 3.00× | 5.00× |
A dependency degrading to 95% failure receives ~2.85× its normal traffic from retries alone — at the exact moment it is least able to serve it. It cannot recover, because the moment it serves anything the backlog surges again.
The fix order matters, and most people get it backwards:
- Retry budget — first and most important. Cap retries at a fraction (10% is a common choice) of base traffic. Amplification is then bounded at 1.1× no matter how bad things get. This is what gRPC's retry throttling and Envoy's retry budgets implement.
- Circuit breaker. Stop trying entirely once failure is established.
- Jitter. Desynchronize the retries you do send.
Jitter alone is popular advice and it is insufficient: perfectly jittered retries still deliver 2.85× load. Only a budget bounds it.
Also: do not retry at every layer. Three layers each retrying 3× is 27 attempts. Retry at one layer — usually the outermost that knows the request is idempotent — and pass failures through elsewhere.
10.2 Circuit breakers
Three states:
- Closed — normal. Count failures.
- Open — failure threshold exceeded. Fail immediately without calling. This is the point: you stop wasting your own capacity, and you stop adding load to a struggling dependency.
- Half-open — after a cooldown, allow a small number of probes. Success → closed. Failure → open again.
The parameters are genuinely hard and worth acknowledging:
- Threshold: an absolute count breaks on low-traffic endpoints (3 failures out of 5 requests is noise). Use a rate over a minimum volume — "50% failures over at least 20 requests in 10 seconds".
- Cooldown: too short and you hammer a recovering dependency; too long and you stay down after it recovers. Exponential with jitter is the usual answer.
- Half-open concurrency: let one probe through, not the full flood, or reopening the circuit re-kills the dependency instantly.
Scope matters more than parameters. One breaker for an entire service means one bad endpoint trips everything. Per-endpoint, and often per-instance, is right — that is how outlier detection ejects a single fail-slow node without failing the whole dependency.
10.3 Load shedding and admission control
When you cannot serve everything, choose what not to serve. Rejecting 10% in 1 ms so the other 90% meet SLO is strictly better than accepting 100% and timing all of them out — in the second case everyone loses and you burned 30 seconds of capacity per doomed request.
Shed by priority, not at random. Health checks and control-plane traffic first; then paid tiers; then best-effort. That requires request classification at the edge, which is a design decision to surface early.
Shed the oldest queued item. Counter-intuitive until you notice that under sustained overload, FIFO serves nothing but requests whose clients have already given up. Serving them is pure waste. Some systems go further and use LIFO under load for exactly this reason.
Deadline propagation is the sophisticated version and it is worth naming: the client sends its deadline, every hop passes the remaining budget downstream, and any service that sees insufficient time remaining fails immediately rather than starting work it cannot finish. gRPC deadlines work this way. It converts wasted capacity into fast failures across the whole call graph.
10.4 Cascading failure and bulkheads
The pattern: service A calls B. B slows down. A's threads block waiting on B. A's pool fills. A now fails every request — including ones that never touch B. A's callers then fill their pools. The failure propagates upward through healthy services.
Bulkheads — from ship compartments — contain it: separate resource pools per dependency, so exhaustion in one cannot starve the others.
thread pool for B: 20 thread pool for C: 20 thread pool for D: 10
B saturating consumes only its 20. Requests to C and D are unaffected.
Related containment:
- Timeouts on everything. An unbounded wait is how the pool fills in the first place. Every network call needs a timeout, and it should be derived from the caller's deadline rather than hard-coded.
- Cellular architecture. Partition the entire stack into independent cells, each serving a slice of users. A failure is contained to its cell, so blast radius is 1/N by construction. This is how AWS structures many services, and it is the strongest available answer to "how do you limit blast radius".
The honest counter-argument, which you should raise yourself: bulkheads reduce pooling efficiency. By M/M/c queueing, one pool of 50 has better tail latency than five pools of 10, because a burst on one dependency can borrow idle capacity. You are trading efficiency for isolation, and that trade is worth naming rather than presenting bulkheads as free.
10.5 The thundering herd after recovery
The failure that happens during recovery, and the one designs usually forget.
A service comes back after a two-hour outage. Every client has been retrying. Every scheduled job is overdue. Every cache is cold. All of it arrives at once, and the service — which just started, with empty caches, cold JITs and empty connection pools — is at its weakest exactly when load is at its highest. It falls over again immediately.
Mitigations, all of which should appear in a good failure section:
- Jittered restarts and reconnects so clients do not arrive in lockstep.
- Rate-limited catch-up: drain the overdue backlog at a bounded rate, and prioritize new work over overdue work — new requests have someone waiting on them.
- Cache warming before accepting full traffic; a shadow-traffic phase is ideal.
- Slow-start / ramped admission: accept 10% of traffic, then 20%, watching health. This is what a load balancer's slow-start does for a newly-added backend.
- Explicit catch-up policy for scheduled work:
run_all,run_latest_only, orskip— a per-job decision the design must expose rather than silently make. Firing 40,000 overdue jobs at once turns one outage into a second, worse one.
Chapter 11: CRDTs, When You Can Avoid Coordination
Consensus is expensive. Sometimes you can skip it entirely.
A CRDT (Conflict-free Replicated Data Type) is a data structure whose merge operation is commutative, associative, and idempotent. Those three properties mean replicas that receive the same updates in any order, any number of times, converge to the same state — with no coordination at all.
Why those three:
- Commutative — order does not matter, so messages can arrive out of order.
- Associative — grouping does not matter, so merges can be batched arbitrarily.
- Idempotent — repeats do not matter, so at-least-once delivery is safe.
Together they mean the merge is a join on a lattice, and convergence is a theorem rather than a hope.
| CRDT | What it does | The trick |
|---|---|---|
| G-Counter | increment-only counter | per-node counts; merge = element-wise max; value = sum |
| PN-Counter | increment/decrement | two G-Counters (P and N); value = sum(P) − sum(N) |
| G-Set | add-only set | merge = union |
| 2P-Set | add and remove once | two G-Sets; removed = in the tombstone set. Cannot re-add |
| LWW-Register | last write wins | timestamp + node id tiebreak. Lossy — a concurrent write is discarded |
| OR-Set | add/remove freely | each add gets a unique tag; remove removes the tags you saw, so a concurrent add survives |
| RGA / Logoot | ordered sequence | the basis of collaborative text editing |
The costs, which are why CRDTs are not the default:
- Tombstones grow. Removed elements must be remembered, or a delayed add resurrects them. Garbage-collecting tombstones safely requires knowing every replica has seen the removal, which is a coordination problem — the one you were trying to avoid.
- Metadata can exceed the data. An OR-Set of small strings may carry more tag bytes than payload.
- Convergence is not correctness. A CRDT guarantees all replicas agree. It does not guarantee they agree on something useful. A CRDT counter for inventory converges — to a number that may be negative, because no replica ever saw the stock run out. Invariants that span replicas still need coordination.
Point 3 is the one to say out loud. It is the honest limit, and it is why CRDTs are excellent for collaborative editing, presence, and counters-with-no-invariant, and wrong for anything with a constraint like "never oversell".
The Twelve Design Prompts
Work these in order. d01 and d02 first and they are not optional. For each, produce the
nine-section template, then attack your own design with the failure catalog, then revise.
| # | Prompt | The two hard parts |
|---|---|---|
| d01 | Fault-tolerant distributed job scheduler | Exactly-once dispatch under scheduler failure · worker liveness, leases, split brain |
| d02 | Distributed versioned KV store | Global version ordering (consensus) · consistent snapshots across shards |
| d03 | Distributed rate limiter | Atomic check-and-decrement · fail-open vs fail-closed |
| d04 | Webhook delivery system | Per-destination isolation · at-least-once + DLQ + replay |
| d05 | Load shedding / admission gateway | Priority classification · deadline propagation |
| d06 | Feature store (online + offline) | Training/serving skew · point-in-time correctness |
| d07 | Log analytics pipeline | Ingest backpressure · index vs query cost |
| d08 | Multi-region metadata store | Cross-region write latency · conflict resolution |
| d09 | Search / retrieval serving | Index freshness vs query latency · shard fan-out tail |
| d10 | Event streaming platform | Consumer group rebalancing · ordering vs parallelism |
| d11 | Distributed lock / coordination service | Consensus · fencing |
| d12 | Multi-tenant control plane | Isolation · fairness under noisy neighbours |
A fully worked example of d01 — all nine sections, with the hostile critique and the revision
— is in designs/d01-job-scheduler.md. Read it after you have
attempted your own, not before.
The Forty Questions
If any answer takes more than fifteen seconds, that is your next study item.
Arithmetic
- State Little's law and give two uses.
- Response-time multiplier at ρ = 0.9? At 0.95?
- Why does real traffic hit the knee earlier than M/M/1 predicts?
- Round trip within a datacenter? Cross-country? Transatlantic?
- Why is "fan out, don't chain" an arithmetic statement?
Failure 6. Name the three legs of a failure analysis. 7. Why is fail-slow worse than fail-stop? 8. How do you detect fail-slow? 9. What is a deliberately accepted failure mode, and why say one out loud? 10. What happens to a service in the first minute after it recovers?
Time 11. Wall clock vs monotonic — which for a lease, and why? 12. What does NTP guarantee? 13. What do Lamport timestamps give you, and what can they not do? 14. What do vector clocks add, and what do they cost? 15. What problem do hybrid logical clocks solve? 16. What is commit-wait and why does Spanner need it?
Leases and fencing 17. Why does a lease expire rather than a lock? 18. Narrate the zombie scenario. 19. Why can't you detect a zombie? 20. What is a fencing token and who must check it? 21. Why doesn't "check your lease before writing" work?
Replication and consensus 22. Sync vs async vs semi-sync — what do you lose on failover? 23. Why does W + R > N work? 24. Name three things a quorum does not give you. 25. What is read repair, and what is anti-entropy? 26. Why are Raft terms a logical clock? 27. What are the two Raft safety rules? 28. Why can't a leader commit a previous term's entry by counting replicas? 29. Name three costs of consensus. 30. Why 3 or 5 nodes, never 4?
Consistency 31. Linearizability vs serializability, in one sentence each. 32. What is strict serializability? 33. State CAP precisely. Now state PACELC. 34. Name three session guarantees and what they cost.
Partitioning and load
35. Why does hash(key) % N fail on resize? What fixes it?
36. Why virtual nodes?
37. How do you find and fix a hot partition?
38. Why is the fix order budget → breaker → jitter, not jitter first?
39. Why shed the oldest queued item?
40. What do bulkheads cost you?
References
Books
- Kleppmann, M. Designing Data-Intensive Applications, 2nd ed. O'Reilly. — Ch. 5 replication, 6 partitioning, 7 transactions and write skew, 8 unreliable clocks and fencing, 9 consistency and consensus. The single best source for this track
- Beyer et al. Site Reliability Engineering. O'Reilly, 2016. — Ch. 21 handling overload, Ch. 22 cascading failures
- Beyer et al. The Site Reliability Workbook. — the practical companion
- Tanenbaum, A. and van Steen, M. Distributed Systems, 4th ed. — the textbook treatment
Papers
- Ongaro, D. and Ousterhout, J. In Search of an Understandable Consensus Algorithm (Raft). USENIX ATC 2014. https://raft.github.io/raft.pdf
- Lamport, L. Time, Clocks, and the Ordering of Events in a Distributed System. CACM, 1978
- Lamport, L. Paxos Made Simple. 2001
- Fischer, Lynch, Paterson. Impossibility of Distributed Consensus with One Faulty Process. JACM 1985 — FLP
- Gilbert, S. and Lynch, N. Brewer's Conjecture and the Feasibility of Consistent, Available, Partition-Tolerant Web Services. 2002 — the CAP proof
- Abadi, D. Consistency Tradeoffs in Modern Distributed Database System Design. IEEE Computer 2012 — PACELC
- DeCandia et al. Dynamo: Amazon's Highly Available Key-value Store. SOSP 2007 — consistent hashing, vector clocks, sloppy quorums
- Corbett et al. Spanner: Google's Globally-Distributed Database. OSDI 2012 — TrueTime, commit-wait
- Burrows, M. The Chubby Lock Service. OSDI 2006 — leases and sequencers in production
- Kulkarni et al. Logical Physical Clocks and Consistent Snapshots in Globally Distributed Databases. OPODIS 2014 — HLC
- Shapiro et al. Conflict-free Replicated Data Types. SSS 2011
- Ports, D. and Grittner, K. Serializable Snapshot Isolation in PostgreSQL. VLDB 2012
- Karger et al. Consistent Hashing and Random Trees. STOC 1997
Engineering writing
- Kleppmann, M. How to do distributed locking. https://martin.kleppmann.com/2016/02/08/how-to-do-distributed-locking.html — the fencing argument and the Redlock critique
- Kingsbury, K. Jepsen analyses. https://jepsen.io/analyses — the best catalog of how real systems actually break
- Brooker, M. Exponential Backoff and Jitter. https://aws.amazon.com/blogs/architecture/exponential-backoff-and-jitter/
- Amazon Builders' Library — Timeouts, retries, and backoff with jitter; Using load shedding to avoid overload; Workload isolation using shuffle-sharding
- Netflix. Performance Under Load (AIMD concurrency limits). https://netflixtechblog.medium.com/performance-under-load-3e6fa9a60581
- Google SRE. Addressing Cascading Failures. https://sre.google/sre-book/addressing-cascading-failures/
In this repo
README.md— Track C drills, failure modes, rubricdesigns/d01-job-scheduler.md— a fully worked design, with critique and revisioncalculators/envelope.py— the arithmetic in Chapter 1, runnable../coding/WARMUP.md— the single-node versions of these primitives../ml-infra/WARMUP.md— the same reasoning applied to GPU serving