"Design a key-value store" is two questions in one. The first is about data placement: how a billion keys spread across machines so you can add capacity without reshuffling everything. The second is about disagreement: what happens when a replica is down during a write, or two clients write the same key at once, or the network splits the cluster in half. The API has two calls. The interview is about everything behind them.
Build a Dynamo-style key-value store: values are opaque blobs looked up by key, there are no joins or secondary indexes, and the store must scale horizontally and stay writable through node failures.
In scope
get,put,delete- Partitioning, replication, failure handling, conflict resolution
- The storage engine on each node, briefly
A Dynamo-style key-value store keeps get, put and delete writable through node failures. At an assumed 1B keys of 1 KB (3 TB with 3 replicas) and 200,000 ops a second, replication turns traffic into 500,000 replica operations, about 25 nodes. Keys are hashed onto a ring with virtual nodes and written to the next 3 distinct nodes, succeeding at W = 2 and reading at R = 2 so R + W > N; each node uses a commit log, memtable and SSTables. The deep dives cover virtual nodes, sloppy quorums with hinted handoff, last write wins versus vector clocks, read repair and Merkle trees, then partitions, gossip, tombstones and hot keys.
Phase 1: Requirements (3 minutes)#
"Is it more important that a write is never rejected, or that a read never returns stale data? Are values small? Do we need transactions across keys?"
"Think shopping cart: never reject a write. Values are small, under 1 MB. No multi-key transactions."
Functional requirements#
put(key, value, context?),get(key)returning value and version context,delete(key)- Per-request consistency level:
ONE,QUORUMorALL - Values up to 1 MB, opaque to the store
Non-functional requirements#
- Always writable: a write succeeds as long as enough nodes are reachable, even if some replicas are down
- Low latency: single-digit milliseconds at p99 for small values (a design target)
- Horizontal scale: add nodes to add capacity, moving only a small share of data
- Durable: 3 copies, on different machines (ideally different racks or zones)
- No single leader: any node can serve any request, so no node is a single point of failure
Phase 2: Estimation (3 minutes)#
- Data: assume 1B keys × ~1 KB ≈ 1 TB raw, ~3 TB with 3 replicas (plus compaction headroom)
- Traffic: assume 200,000 ops/s at peak, half reads and half writes
- Replication multiplies the work. With N = 3, W = 2, R = 2:
- each write is sent to all 3 replicas: 100,000 × 3 = 300,000 replica writes/s
- each read asks 2 replicas: 100,000 × 2 = 200,000 replica reads/s
- total: 500,000 replica operations/s
- Assume a node serves ~20,000 replica ops/s comfortably on SSDs: 500,000 ÷ 20,000 = 25 nodes
- Storage per node: 3 TB ÷ 25 ≈ 120 GB, so throughput, not disk, sets the node count
"The interesting number is the 2.5x: every client operation becomes about two and a half operations inside the cluster. That's the price of three copies."
Phase 3: API (2 minutes)#
PUT /kv/cart:8812?consistency=QUORUM
{ "value": "<bytes>", "context": "<opaque version from the last read>" }
→ 200 { "context": "<new version>" }
GET /kv/cart:8812?consistency=QUORUM
→ 200 { "values": ["<bytes>"], "context": "<version>" }
(more than one value means concurrent versions: the caller merges and writes back)
DELETE /kv/cart:8812 → 200 (writes a tombstone, removed later)The context is the version information (a vector clock, see deep dive 2) that lets the store tell an update from a conflicting concurrent write.
Phase 4: High-level design (4 minutes)#
%% title: Leaderless key-value store: any node coordinates
flowchart TD
APP@{ icon: "sd:servers", label: "App servers<br/>client library" } --> CO["Coordinator<br/>(whichever node got the request)<br/>hash(key) → preference list"]:::core
subgraph RING["Hash ring · 25 nodes × virtual nodes"]
A@{ icon: "sd:database", label: "Node A · replica 1" }
B@{ icon: "sd:database", label: "Node B · replica 2" }
C@{ icon: "sd:database", label: "Node C · replica 3" }
D@{ icon: "sd:database", label: "Node D · next on the ring" }
end
CO -->|"write to 3, wait for W = 2"| A
CO --> B
CO --> C
CO -.->|"if C is down: hinted handoff"| D
G@{ icon: "sd:network", label: "Gossip<br/>membership + failure detection" } -.- RING
class APP client
class A,B,C,D store
class G service- Hash ring: each key is hashed to a position on a ring; nodes own ranges of it. The preference list for a key is the next 3 distinct physical nodes clockwise.
- Coordinator: any node can take a request. It finds the preference list, sends the operation to the replicas and waits for the required number of answers. A smart client library can send requests straight to a replica and skip a hop.
- Gossip: each node periodically exchanges membership and heartbeat state with a few random peers, so every node learns the ring and who looks dead without a central master.
- Storage engine on each node: a log-structured merge tree. A write appends to a commit log (durability) and an in-memory memtable; full memtables are flushed as immutable sorted files (SSTables), which compaction merges in the background. Reads check the memtable, then SSTables, skipping files whose Bloom filter says the key is absent.
Phase 5: Deep dives (5 minutes)#
Deep dive 1: Placement and quorums#
"Why a ring, and why not just hash(key) % 25?"
"With modulo, adding a 26th node changes the owner of almost every key. On a ring, a new node takes over only the ranges next to its positions, roughly 1/26 of the data."
- Hash keys and nodes onto one ring (for example a 64-bit hash space). A key belongs to the first node clockwise from its position.
- Give each physical node many positions, say 128 virtual nodes. With one position per node, arcs are uneven and a new node unloads just one neighbour; with 128, load evens out and a new node takes small slices from everyone.
- Replicate to the next N = 3 distinct physical nodes clockwise, skipping extra virtual nodes of a machine already chosen, and preferably in different racks or zones.
- Rebalancing streams only the moved ranges from their old owners, in the background.
- N = 3 replicas. A write waits for W = 2 acknowledgements, a read waits for R = 2 answers.
- R + W = 4 > 3, so any read set and any write set share at least one node: the read sees the latest acknowledged write and returns the newest version.
- Tune per request:
W = 1for fast, lossy writes (metrics),R = 1for fast reads that can be stale,QUORUMfor both by default,ALLwhen you must, at the cost of availability.
Deep dive 2: Failures and disagreement#
"Replica C is down during a write. And separately: two phones add items to the same cart at the same moment. What happens?"
- Sloppy quorum: the coordinator writes to A, B and the next healthy node D instead of C. W = 2 is still met, so the write succeeds.
- Hinted handoff: D stores the copy with a hint, "this belongs to C". When gossip says C is back, D sends it over and deletes its copy.
- Read repair: when a quorum read sees replicas disagree, the coordinator returns the newest version and writes it back to the stale replica.
- Anti-entropy: in the background, replicas compare Merkle trees (hash trees over key ranges). Matching roots mean a range is identical; a mismatch is narrowed down level by level, so only the differing keys are transferred.
Last write wins
- Each version carries a timestamp; the highest wins
- Simple, one value per key
- Silently drops a concurrent write, and clock skew can make an older write "win"
Vector clocks
- Each version carries a counter per writer node
- Detects when two versions are concurrent rather than one following the other
- Returns both versions; the client merges (union of two carts) and writes back
"For a shopping cart I'd use vector clocks and let the client merge, because losing an item is worse than showing a merge. For a profile photo URL, last write wins is fine and far simpler."
Quorums and failures
Q1/2 N = 3. Which setting guarantees a read overlaps the latest acknowledged write?
Answer & explanation
Answer: C
R + W must exceed N. With W = 2 and R = 2 (sum 4 > 3), every read set shares at least one node with every write set.
Q2/2 Replica C is down during a write with W = 2. How does the store stay writable?
Answer & explanation
Answer: B
This is a sloppy quorum with hinted handoff: D holds C's copy and delivers it when gossip reports C is back.
Phase 6: Trade-offs and failure modes (2 minutes)#
Network partition
With sloppy quorums, both sides keep accepting writes (availability) and you reconcile later (vector clocks, read repair). With strict quorums, the minority side rejects writes (consistency). This is the CAP choice, made per request through the consistency level.
Deletes come back
A delete is a tombstone, not an erase. If tombstones are purged before every replica has seen them, an old replica can resurrect the value through repair. Keep tombstones longer than the longest expected outage, then let compaction drop them.
Hot keys
One very popular key lands on 3 nodes no matter how big the cluster is. Cache it in front of the store, or split it into sub-keys the application combines.
Compaction and repair spikes
Compaction and Merkle repair compete with live traffic for disk. Throttle them, and schedule full repairs off-peak.
Adding a node
With virtual nodes, a new node pulls small ranges from many peers instead of hammering one. Stream them gradually so latency stays flat.
Phase 7: Go deeper in the course#
- Consistent hashing, virtual nodes, rebalancing: Week 1 · Day 1: Partitioning Deep-Dive and the Lab: Consistent hashing ring with virtual nodes
- Quorums, hinted handoff, read repair, failover: Week 1 · Day 2: Replication Trade-offs
- Strong, eventual and read-your-writes consistency: Week 5 · Day 1: Consistency Models
- Last-write-wins, vector clocks, CRDTs, tombstones: Week 5 · Day 4: Conflict Resolution
- Leader-based designs, for contrast: Week 5 · Day 5: Leader Election and Coordination
- CAP, Bloom filters and the vocabulary: Week 0 · Part 6: Terminology and Theory
- Hot keys: Week 1 · Day 4: Hot Keys and Skew
- An object store built on the same ideas: Side quest: AWS S3
Final check#
Final check · Key-value store
Q1/4 200,000 client ops a second, half reads and half writes, with N = 3, W = 2, R = 2. How many replica operations does the cluster do?
Answer & explanation
Answer: C
Writes are sent to all 3 replicas (100,000 × 3) and reads ask 2 (100,000 × 2). Replication multiplies the work, which is why throughput sets the node count.
Q2/4 Why give each physical node many virtual nodes on the ring?
Answer & explanation
Answer: B
With one position per node, arcs are uneven and adding a node relieves only one neighbour. Many positions spread both load and rebalancing.
Q3/4 Two clients add different items to the same cart concurrently. Which approach keeps both items?
Answer & explanation
Answer: D
Vector clocks detect that the versions are concurrent, so the store returns both and the client writes back their union. Last write wins would silently drop one.
Q4/4 Why can a deleted value come back to life?
Answer & explanation
Answer: A
A delete is a tombstone. If it disappears before a lagging replica learns of it, repair copies the old value back. Keep tombstones longer than the longest outage.
New lessons by email
New lessons, side quests and case files as they ship, at most once a week. No spam, unsubscribe anytime.
Discussion · …
Public: everyone reading this lesson sees comments. Be kind and keep it on the material.