Design a distributed key value store
This is the question where every building block shows up at once. Consistent hashing decides where data lives, quorums decide what a read returns, replication decides what survives, and the interesting parts are all about what happens when a node is missing.
If you can do this one, most of the others become assembly.
A node is down when a write arrives, then comes back an hour later holding stale data. How does it find out it is stale, without comparing every key it owns?
What you are building
- get(key) and put(key, value). That is the entire API, and keeping it that small is what makes the rest possible.
- Data spread across many nodes with configurable replication. No node holds everything and no node is special.
- Stay writable while nodes are down. This is the design goal that forces almost every other decision.
- Repair replicas that fell behind. Both quickly on read and slowly in the background.
- Values up to about 1MB. Larger belongs in object storage with the key here.
- Single digit millisecond reads at p99 inside one region.
- No single point of failure and no leader to elect.
- Adding a node must not require moving most of the data.
- Range scans and secondary indexes. Get and put only, and saying so early keeps the design honest.
- Multi key transactions. A store built for availability cannot offer them cheaply, and pretending otherwise is where these designs go wrong.
- Cross region replication. Same ideas, much longer round trips, and enough material for its own page.
The numbers
At 8 nodes, a node being down is not an incident, it is Tuesday. That is the whole argument for a design with no leader and no special node: at this size something is always restarting, and the system has to treat that as normal rather than as a failure to recover from.
Where a key lives
Consistent hashing, with virtual nodes. A key hashes onto the ring and belongs to the first node clockwise, and its replicas are the next N distinct physical nodes after that.
Add and drop nodes above. The counter is the reason this technique is here: at a few hundred nodes, adding one moves a fraction of a percent of the data, and every node can compute the answer locally with no lookup service in the request path.
Replication then falls out for free. Walk clockwise from the key and take the next N distinct physical nodes, skipping virtual nodes belonging to a machine already in the list. That skip matters: without it, three replicas can land on one machine and the replication factor becomes a lie.
What a read returns
N, R and W are the dials the operator gets. The store does not decide consistency, the caller does, per operation, and that is the feature.
The common settings say something real. W=1, R=1 is a cache: fast, always writable,
frequently stale. W=N, R=1 makes reads cheap and writes fragile, since one slow replica
stalls every write. W=2, R=2 with N=3 is the workhorse, because it survives one node
being down on either side while still overlapping.
R + W > N guarantees a read sees the latest completed write. It says nothing about two clients writing the same key at the same moment. Both can succeed, and now there are two versions with no ordering between them. That is a separate problem, and pretending the quorum solved it is the most common mistake in this question.
Deep dive one: two writers, one key
The alternative is last write wins: attach a timestamp, keep the higher one, discard the other. It is simple, it is what many deployments use, and it silently loses data. Worse, under clock skew it can lose the newer write. Say out loud which one you are choosing and what it costs, because both are defensible and only one of them is defensible silently.
Deep dive two: nodes are always missing
The Merkle tree is the answer to the question at the top of this page. A node returning after an hour does not compare every key it owns. It compares a root hash with its peers, and only descends into the parts of the tree that disagree.
The storage engine
Writes go to a commit log and an in memory table, and when that table fills it is flushed to an immutable sorted file. Files are merged in the background. Reads check memory, then files newest to oldest, with a bloom filter per file so most files are skipped without being opened.
This is an LSM tree, and the reason to pick it over a B-tree here is that every write is an append. No random disk writes, no page splits on the write path, and immutable files are trivially safe to copy for repair or for streaming to a new node.
| commit log | append only | Every write lands here first. This is what makes a crash survivable. | |
| memtable | sorted, in memory | The current writes. Flushed when it hits a size threshold. | |
| sorted files | immutable | IDX | Never modified after writing, which makes them safe to stream elsewhere while live. |
| bloom filter | per file | Answers "definitely not in this file" so a read skips most files without touching disk. | |
| merkle tree | per key range | Recomputed as files merge. This is what makes anti entropy cheap. |
Break it
The API
{
"value": "...",
"version": "n1:4,n3:2"
}Trade-offs
| Choice | What you gain | What you pay | Pick it when |
|---|---|---|---|
| Leaderless with quorums | No election, no failover pause, and writes continue while nodes are restarting. | Concurrent writes produce siblings someone has to reconcile, and there are no multi key transactions. | Availability matters more than ordering and the data model has a natural merge. |
| Version vectors over last write wins | Concurrent writes are detected rather than silently discarded. | Applications must handle siblings, and the vector grows with the number of writing nodes. | Losing a write would be noticed. Last write wins is fine for a value that is simply overwritten. |
| LSM tree over B-tree | Every write is sequential, and immutable files are cheap to stream to a recovering node. | Reads may touch several files, and background compaction consumes real resources. | Write heavy workloads, which is most of what these stores are chosen for. |
| Merkle trees for anti entropy | Comparing two replicas costs one hash when they agree, and traffic proportional to the difference when they do not. | Trees to maintain and recompute as files compact. | Any replicated store where replicas can drift, which is all of them. |
Interview replay
Checkpoint
1. How does a returning node discover which keys are stale without comparing them all?
2. N = 3, W = 2, R = 2, and two clients write the same key concurrently. What does the store do?
3. Why does a delete write a tombstone rather than removing the row?
Consistent hashing with virtual nodes places keys, and replicas are the next N distinct physical nodes clockwise, so every node computes placement locally with no lookup service in the path. Consistency is per operation through N, R and W: with R plus W greater than N every read set overlaps every write set, so a read sees the latest completed write. What that does not give me is ordering between concurrent writes, so I would use version vectors, keep both as siblings, and let the application merge, rather than last write wins which silently drops a write. Availability during failures comes from hinted handoff, where a write meant for a down node goes to a neighbour tagged for later delivery. Repair is two mechanisms: read repair fixes popular keys for free because reads already contact several replicas, and Merkle tree anti entropy fixes the keys nobody reads by comparing root hashes and descending only where they differ. Storage is an LSM tree, so every write is sequential and the immutable files are cheap to stream to a recovering node.
