Library
LearnSystem designChapter 16
23 min read

Design a distributed key value store

SeniorReplicationShardingConsistencyStorage and indexesUpdated

Read firstConsistent hashingReplication and quorums

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.

Step 1: Understand the problem#

Half of this question is deciding what you are not going to support. Every "no" below buys you something in the design, so make them out loud and early.

You askThey sayWhat it settles
What operations does the store have to support?get and put by key. No range scans, no secondary indexes.Everything. Get and put by key is what makes hashing keys across nodes legal. Add range scans and you need ordered partitioning and a very different design.
Multi key transactions?No.No coordinator holding locks across nodes, which is what lets any node serve any request. Say this early, because a store built for availability cannot offer transactions cheaply.
When a node is down, do writes fail or succeed?Succeed. The store must stay writable.Availability over consistency, so no leader and no election pause. It also means concurrent writes can conflict, and something has to reconcile them later.
How big can a value be?Up to about 1MB.Values are passed around whole, so replication traffic is predictable. Anything bigger goes to object storage with the key kept here.
How many nodes, and how often do they change?Hundreds, and something is always restarting.Consistent hashing rather than modulo, so adding a node moves a fraction of a percent of the data. At this size a missing node is normal operation, not an incident.
One region or many?One region for now.Round trips stay under a millisecond, so a quorum of two out of three is cheap. Across regions the same quorum costs 150ms and the whole latency argument changes.

What you are building#

In scope
  • 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.
The numbers you commit to
  • 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.
Cut, and say so out loud
  • 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.

Back of the envelope#

Cluster sizing
20 billion
1000 bytes
3
Logical data20B x 1060B = 21 TB
With 3 replicas64 TB
Nodes at 8TB each64 / 8 = 8
Keys moved when one node joins1 / 8 = 12.50%
8 nodes
and one of them is always doing something unusual

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.

Step 2: Propose the high level design#

The API#

GET/v1/keys/{key}?r=2
returns 200 { value, version, siblings: [] }
Why: The version comes back with the value and must be passed to the next write. Without it the store cannot tell an update from a blind overwrite, and every write becomes a potential lost update.
PUT/v1/keys/{key}?w=2
{
  "value": "...",
  "version": "n1:4,n3:2"
}
returns 200 or 409 with siblings
Why: R and W are query parameters, so consistency is chosen per operation. A session token can be written with w=1 and a billing record with w=3 in the same application, which is the main reason to run a store like this.
DELETE/v1/keys/{key}
returns 204
Why: A delete writes a tombstone rather than removing anything. Without it, a replica that was down would resurrect the key when it came back and compared trees. Tombstones are kept past the maximum expected downtime, then collected.

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.

1
Copies of each key
Add or drop a node and this counts how many keys had to move.
Share of the ringeven is 20%
n0
26%
n1
4%
n2
17%
n3
25%
n4
28%
Pick a key to see its walk

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 works out where a key lives locally, with no lookup service anywhere 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#

Preset
3
2
2
W + R - N = 1 overlap, so reads see the latest write
Every read set shares at least 1 node with every write set, so a read always touches a replica that has the newest value. Writes survive 1 node failure, reads survive 1.

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.

Quorum does not order concurrent writes

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.

The whole system on one whiteboard#

Figure 1 is the entire store. There is no leader in it, no config service in the request path, and no box that every request has to pass through. That is not minimalism, it is the design goal: at a few hundred nodes something is always restarting, so nothing is allowed to be special.

Figure 1. Any node can coordinate any request. It hashes the key locally, talks to the three replicas that own it, and waits for W of them. The three dashed boxes are the repair machinery, and none of it is on the request path.

Walking Figure 1:

  1. The client connects to whichever node it likes. There is no routing tier to keep alive and no leader to find.
  2. That node becomes the coordinator for this request. It hashes the key against its own copy of the ring, which it maintains from gossip, so placement costs a hash and no network call.
  3. It sends the operation to the N nodes that own the key.
  4. It waits for W acknowledgements on a write, or R responses on a read, and answers the client. It does not wait for the slowest replica, which is where the tail latency of this design comes from.
  5. Each replica appends to its own commit log and memtable. Nothing is updated in place.
  6. Repair runs beside all of this: hints for nodes that were absent, read repair on the values a read already fetched, and Merkle trees for the keys nobody reads.
Your answer

A node is down when a write arrives, then comes back an hour later holding stale data. How does it find out which keys are stale, without comparing every key it owns?

Step 3: Design deep dive#

Two writers, one key#

1/5 Client A writes, and the write is tagged with a version. A plain timestamp is the tempting choice and the wrong one, because clocks on different machines disagree.

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.

Nodes are always missing#

Repair mechanism
A replica is down, so the coordinator writes to the next node along and tags it as a hint for the node that was missing. When that node returns, the hint is delivered and deleted. Writes never fail because a replica is restarting, which is the point, and the cost is a window where a hint holder is carrying data it does not own.

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.

on disk layoutlog structured, per node
commit logappend onlyEvery write lands here first. This is what makes a crash survivable.
memtablesorted, in memoryThe current writes. Flushed when it hits a size threshold.
sorted filesimmutableIDXNever modified after writing, which makes them safe to stream elsewhere while live.
bloom filterper fileAnswers "definitely not in this file" so a read skips most files without touching disk.
merkle treeper key rangeRecomputed as files merge. This is what makes anti entropy cheap.
Sample row
put(cart:7) -> commit log -> memtable -> flush -> sorted file 0041
Compaction is the cost. Merging files consumes disk and CPU in the background, and a cluster that falls behind on compaction gets slower reads because a lookup has to check more files. It is the operational thing that actually goes wrong with these stores.

Break it#

Cluster health
healthy
healthyone node downtwo nodes downpartitionrepaired
Healthy. N = 3, W = 2, R = 2. Every read overlaps every write, latency is the second fastest replica rather than the slowest, and nothing is coordinating anything.
Read every step as text
  1. healthy · Healthy. N = 3, W = 2, R = 2. Every read overlaps every write, latency is the second fastest replica rather than the slowest, and nothing is coordinating anything.
  2. one node down · Healthy. A replica restarts. W = 2 is still satisfiable by the other two, so writes continue with no error, and hinted handoff parks the missed writes on a neighbour. This is not degraded mode, it is the mode the system was designed for.
  3. two nodes down · Degraded. Two of three replicas for some key range are gone. W = 2 cannot be met, so writes to those keys fail. Correct behaviour, and worth saying out loud: the store refuses rather than accepting a write it cannot make durable.
  4. partition · Down. The cluster splits. Each side can still reach some replicas and both keep accepting writes to the same keys, which is exactly what an available store is supposed to do. The bill arrives at the merge as sibling versions the application has to reconcile.
  5. repaired · Healthy. The partition heals. Hints are delivered, read repair fixes the popular keys immediately, and anti entropy walks the Merkle trees to catch everything nobody has read. Conflicts surface as siblings on the next read of an affected key.

Trade-offs#

ChoiceWhat you gainWhat you payPick it when
Leaderless with quorumsNo 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 winsConcurrent 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-treeEvery 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 entropyComparing 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#

Interviewer
How do you decide which nodes hold a key?
The opener, and the answer should take fifteen seconds.
You
Consistent hashing with virtual nodes. The key hashes onto a ring and belongs to the first node clockwise, and replicas are the next N distinct physical nodes after it, skipping additional virtual nodes of a machine already in the list. Every node can compute this locally, so there is no lookup service in the request path, and adding a node moves roughly one Nth of the data instead of nearly all of it.
The word "distinct" is doing real work. Without the skip, replicas can land on one machine.
Interviewer
Two clients write the same key at the same time. Who wins?
The question that separates people who understand quorums from people who have heard of them.
You
Neither, and that is deliberate. R plus W greater than N tells me a read sees the latest completed write, but it says nothing about two writes that never saw each other. With version vectors the store can tell they are concurrent rather than ordered, so it keeps both and returns siblings on the next read. The application merges, because only it knows a cart is a union. The alternative is last write wins, which is simpler and silently discards one of them.
Names the guarantee, names its limit, gives both options and what each costs. This is the strongest answer in the round.
Interviewer
A node is down for an hour. What happens to writes meant for it?
Straight at hinted handoff.
You
They go to the next node along, tagged as hints for the node that was missing, so writes never fail because a replica is restarting. When it returns, the hints are handed over and deleted. That covers writes during the outage but not drift from before it, which is what anti entropy is for.
Ends by naming the gap the mechanism does not cover, which invites the next question and shows the boundary is understood.
Interviewer
So how does it catch up on everything else without comparing every key?
The follow up the last answer set up.
You
Merkle trees per key range. Two replicas compare root hashes: equal means identical and we are done, which is one comparison for millions of keys. Different means descend and compare children, so traffic is proportional to how much actually differs rather than to how much data exists. Read repair handles popular keys immediately as a side effect of reads already contacting R replicas, and anti entropy is what fixes the keys nobody ever reads.
Explains the cost model rather than just naming the structure, and ties the two repair paths together.

Checkpoint#

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?

Worth memorising
  • N=3, W=2, R=2. R + W > N, so a read always overlaps the last completed write.
  • A few hundred nodes means adding one moves a fraction of a percent of the keys. That is the point of the ring.
  • Quorum says nothing about two concurrent writes. That is what version vectors are for.
  • One Merkle root comparison proves millions of keys are identical. Repair traffic follows the difference, not the data size.
Say this in 60 seconds

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.

Related: Consistent hashingRelated: Replication and quorumsRelated: CAP theorem