Back to System design
Design a Distributed Key-Value Store (Dynamo-style)
hardA key-value store saves a value under a
key and gives it back when you ask. get("user:42") returns that user's data. There are no tables, no
joins, and no queries beyond "fetch this key".
ScaleLinear scale with nodes; ms-level p99
StoragePB-scale across cluster; replication factor × dataset
Key Concepts
1
Spreading one across many machines is what the other system design questions quietly assume you have.
2
Deciding which machine holds a key. Every key is put through a hash function, which turns text into a
number. That number decides which machine stores it.
Deciding which machine holds a key.
3
The naive rule — number modulo how many machines — falls apart the moment a machine is added or removed,
because nearly every key then maps somewhere else. Consistent hashing fixes that; the separate page on it
explains how.
4
Copies, so one dead machine does not lose data. Each key is stored on more than one machine. Three is
the usual number. These copies are called replicas, and the rule for placing them is simple: walk around
the ring from the key's position and use the next three different machines.
Copies, so one dead machine does not lose data.
5
Now the interesting part: how many copies must answer?
Now the interesting part: how many copies must answer?
6
Call the number of copies N. When writing, wait for W of them to confirm. When reading, ask R
of them and take the newest answer.
NWR
7
The rule is that R + W must be greater than N. Here is why it works, with N=3 and W=R=2.
R + W must be greater than N
8
Machines: A B C
Write goes to A B (W=2, C is behind or down)
Read asks B C (R=2)
9
B is in both sets.
10
The write reached two machines. The read asked two machines. With only three machines in total, those two
groups must share at least one — there are not enough machines for them to avoid each other. That shared
machine has the new value, so the read cannot miss it.
11
If you set W=1 and R=1, the sets can miss each other entirely, and a read can return the old value.
12
That is the entire rule, and it is why the settings are per request. A payment uses W=2, R=2. A profile
picture can use W=1, R=1 and be faster.
13
When a machine is down during a write. You do not fail the write. You send it to the next healthy
machine on the ring with a sticky note attached: "this really belongs to C". That note is called a hinted
handoff. When C comes back, the note is delivered and then thrown away.
When a machine is down during a write.
14
When machines have been apart for a long time. After a long outage, two machines may hold different
data for thousands of keys. Comparing them key by key would take forever.
When machines have been apart for a long time.
15
A Merkle tree makes that cheap. Hash each key's value. Then hash those hashes in pairs, and keep hashing
the results in pairs, until one hash is left at the top:
16
top hash <- compare this first
/ \
h(AB) h(CD)
/ \ / \
h(A) h(B) h(C) h(D) <- one hash per chunk of keys
17
Two machines compare the top hash. If it matches, every key underneath matches, and they are done in one
message. If it differs, they compare the two below it, and follow only the branch that disagrees. A
million keys are compared in a handful of messages instead of a million.
18
When two people write at once. During a network split, two clients write different values to the same
key on different machines. Neither write is wrong. Now what?
When two people write at once.
19
The simple answer is last-write-wins: keep the one with the later timestamp. It is easy, and it silently
throws away the other write. It also trusts clocks on different machines to agree, which they do not.
20
The careful answer is a vector clock. Each machine keeps a counter, and the value carries all the counters
it has seen:
21
Alice writes on A -> value "cart: [book]" clock {A:1}
Bob writes on B -> value "cart: [pen]" clock {B:1}
22
Neither clock is bigger than the other in every position, so the system knows these two happened
independently rather than one after the other. It cannot merge them itself — only the application knows
that two shopping carts merge by adding the items together — so it hands both back and asks.
23
That is exactly what Amazon's original design did, and the shopping cart is the example they used.
24
How the data is stored on disk. Writes go to an append-only file and an in-memory table, which is fast
because appending never seeks. Periodically those are merged into sorted files in the background.
How the data is stored on disk.
25
Reads may have to check several files, so each file keeps a small filter that can say "definitely not
here" without touching the disk. This shape is called an LSM tree, and it is chosen whenever writes
outnumber reads.
26
What the interviewer is probing.
What the interviewer is probing.
27
1. "How do you decide which machine holds a key?" Probing: placement and what happens when the
machine count changes. Stalls: "Hash the key modulo the number of machines." Moves up: consistent
hashing, and replicas as the next few distinct machines around the ring.
1. "How do you decide which machine holds a key?"
28
2. "How do you make reads consistent?" Probing: the quorum rule and why it works. Stalls: "Write
to all the copies." Moves up:
R + W > N, explained by the overlap, with N=3, W=2, R=2 as the default.2. "How do you make reads consistent?"R + W > N
29
3. "A machine is down when a write arrives." Probing: whether failure is part of the design.
Stalls: "Return an error." Moves up: store it elsewhere with a note saying where it belongs, and
deliver it when the machine returns.
3. "A machine is down when a write arrives."
30
4. "Two clients wrote the same key during a network split." Probing: honesty about conflicts.
Stalls: "Last write wins." Moves up: that throws away a real write and trusts clocks; vector clocks
detect that the writes were independent and hand both back.
4. "Two clients wrote the same key during a network split."
31
5. "How do two machines catch up after a long outage?" Probing: whether you know the cheap method.
Stalls: "Copy everything across." Moves up: compare a tree of hashes from the top and follow only the
branches that differ.
5. "How do two machines catch up after a long outage?"
High-level design
Ring: consistent hashing with vnodes (256 per physical node typical).
Replicas: next N distinct physical nodes clockwise from primary.
Coordinator: any node receives request, fans out to replicas.
Quorum: wait for W writes / R reads; respond.
Failure: sloppy quorum + hinted handoff during partition; anti-entropy reconciles.
Conflict: LWW (simple) or vector clocks (precise) or CRDTs (automatic merge).
Components
- Coordinator service (any node).
- Local storage engine — LSM-tree (RocksDB, Cassandra's SSTable) for write-heavy.
- Gossip / failure detector (SWIM, phi-accrual).
- Hinted handoff queue.
- Anti-entropy worker (Merkle tree compare).
- Read repair on detected mismatch.
- Compaction (LSM background).
- Client SDK with smart routing (knows the ring; sends directly to coordinator).
Quorum configurations
N = replicas, W = write acks required, R = read responses required
N=3 W=2 R=2 strongly consistent, tolerates 1 node down
write latency = 2nd fastest replica
THE DEFAULT N=3 W=3 R=1 fast reads, no write tolerance (one node down =
writes fail) N=3 W=1 R=1 fastest, eventually consistent
data loss window if the single acking node dies
before replicating N=3 W=1 R=3 fast writes, reads must reach every replica --
fragile under failure N=5 W=3 R=3 tolerates 2 failures, strongly consistent,
higher latency (3rd fastest of 5) Latency intuition
Waiting for W acks means waiting for the Wth FASTEST replica, so
W=2 of 3 is barely slower than W=1 in a healthy cluster and far
more durable. This is the argument for the default.Merkle tree repair, worked
Keys per node range 1,000,000 Naive comparison 1,000,000 key exchanges
Merkle tree over the range
Leaves (one per 1,000 keys) 1,000
Tree depth (binary) ~10 levels Comparison when ranges are IDENTICAL
Exchange root hashes = 1 round trip
Result: no divergence, done. Comparison when ONE key differs
Root differs -> descend
~10 levels, 2 hashes each = ~20 hashes exchanged
Leaf identified -> transfer that 1,000-key bucket
Total = ~10 round trips + 1 bucket Cost of maintaining the tree
Rebuild on flush / compaction; update leaf hashes on write.
Memory: 1,000 leaves x 32 B = 32 KB per range Why this matters
Anti-entropy runs continuously in the background across every
replica pair. Making the healthy case one round trip is what
makes continuous repair affordable.Conflict resolution compared
Last-write-wins (timestamp)
Storage: one value + timestamp.
Behaviour: the later timestamp wins; the other write vanishes.
Risk: clock skew decides correctness. A node 200 ms ahead wins
arguments it should lose.
Use when: writes are idempotent or loss is tolerable (a cache, a
presence flag). Vector clocks
Storage: value + a counter per node that has written it.
Behaviour: detects whether version A descends from B or they are
concurrent. Concurrent versions are BOTH returned.
Cost: the clock grows with the number of writers; needs pruning.
Use when: silently losing a write is unacceptable. Application merge (what Dynamo actually does)
The store returns siblings; the application merges by its own
rules -- union for a shopping cart, max for a counter, manual
choice for a document.
This is the honest answer: the datastore cannot know that two
carts merge by union, so it must not pretend to. CRDTs
Types whose merge is mathematically defined (counters, sets,
registers). The merge always converges without application code.
Cost: you must model your data as a CRDT from the start.