Library
34 min read

Design a distributed key value store

A key value store is the simplest database there is. You give it a name (the key, like cart:7) and some data (the value, like the items in that cart), and later you ask for the data back by name. That is the whole interface: put and get.

Building one on a single machine is a weekend project. Building one that spreads billions of keys across hundreds of machines, keeps working while some of them are down, and never quietly loses a write is this question. Every HLD building block shows up in it: consistent hashing decides where a key lives, quorums decide what a read returns, replication decides what survives a crash. Watch one write go through.

ClientCoordinatorRing + vnodesReplica 1Replica 2Replica 3Commit log

Ask any node. The client sends put(cart:7) to whichever node it likes. There is no leader to find. That node becomes the coordinator for this one request.

Three words come up all through this page, so here they are in plain terms:

  • A node is one machine in the cluster.
  • A replica is one copy of a key. With three replicas, three different nodes hold the same key, so losing one machine loses nothing.
  • Quorum means "enough of them". Instead of waiting for every replica, a write waits for a set number of them to say yes.

If you can do this one, most of the other design questions 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 say 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. Because we only ever look up one exact key, we are free to scatter keys across nodes by hashing them. Range scans ("all keys from cart:1 to cart:9") need keys kept in order, which is a very different design.
Do we need transactions across several keys?No.No node ever has to lock keys on other nodes, so any node can serve any request. A store built to stay available cannot offer multi key transactions cheaply, so say this early.
When a node is down, do writes fail or succeed?Succeed. The store must stay writable.Availability over consistency, so no leader and no pause to elect a new one. It also means two writes to the same key can conflict, and something has to sort them out later.
How big can a value be?Up to about 1MB.Values are copied around whole, so replication traffic is predictable. Anything bigger goes to object storage, with only its key kept here.
How many nodes, and how often do they change?Hundreds, and something is always restarting.Consistent hashing, not 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 between nodes stay under a millisecond, so waiting for two of three replicas is cheap. Across regions the same wait costs 150ms and the latency argument changes completely.

What you are building#

Functional requirements
What the system does.
  • 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 a configurable number of copies. No node holds everything and no node is special.
  • Stay writable while nodes are down. This one goal forces almost every other decision.
  • Repair copies that fell behind. Quickly when a key is read, and slowly in the background for keys nobody reads.
Non-functional requirements
How well it has to do it. 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.
Out of scope
What you 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.

"p99" means the 99th percentile: 99 out of every 100 reads must be at least this fast.

Back of the envelope#

Three inputs. The last two rows are the ones that shape the design. (New to this kind of estimate? The back of the envelope chapter shows the method.)

Cluster sizing
20 billion
1000 bytes
3
Data, one copy20B x 1060B = 21 TB
With 3 copies64 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: something is always restarting, so the system has to treat that as normal, not as a failure to recover from.

The 60 bytes added to each value is a rough allowance for the key and the bookkeeping stored next to it.

Step 2: Propose the high level design#

The core entities
  • Key and valueOpaque bytes. The store never looks inside the value.
  • Replica setThe next N distinct physical nodes clockwise from the key on the ring. With N = 3, three copies.
  • Version vectorOne counter per node that wrote the key. When neither of two vectors is newer, the writes were concurrent and both are kept as siblings.
  • HintA write held by a neighbour for a node that was down, handed over when it comes back.
Never breaks: An acknowledged write is never silently dropped, even when two clients write the same key at once.

The API#

Three calls. The interesting part is what travels with them.

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 instead of removing anything. Without it, a replica that was down would bring the key back to life when it came back and compared notes. Tombstones are kept longer than the longest expected downtime, then cleaned up.

A tombstone is a small record that says "this key was deleted at this version". It sounds wasteful to store a note about nothing, and the third checkpoint question at the end shows why it is needed.

Where a key lives#

With hundreds of nodes, every node needs to answer "which machines hold cart:7?" on its own, in microseconds, and the answer must barely change when a machine joins or leaves.

How do you pick the nodes that hold a key?

  1. First try
    hash(key) mod number of nodes
    Hash the key to a number, divide by the node count, keep the remainder. With 4 nodes, a key that hashes to 10 lives on node 2. Every node can compute it, and it spreads keys evenly.
    Where it breaks: Add a fifth node and the same key now lands on node 0. Almost every key changes home, so adding one machine means moving nearly all the data.
  2. Better
    A hash ring
    Put the nodes on a circle by hashing their names. A key hashes onto the same circle and belongs to the first node clockwise from it. Adding a node only takes over the arc just before it, so only those keys move.
    Where it breaks: With a few nodes the arcs are very uneven, so one machine can own twice its share. And if replicas are simply "the next nodes clockwise", two copies can end up on one machine.
  3. What you ship
    Ring with virtual nodes, replicas on distinct machines
    Each machine is placed on the ring many times (virtual nodes), which evens out the arcs. A key is stored on the first N distinct physical machines clockwise, skipping extra virtual nodes of a machine already chosen.

Try it. Add and drop nodes and watch the counter of keys that move.

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

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. The consistent hashing chapter covers the ring from scratch, and why virtual nodes are not optional shows how lopsided it gets without them.

The "skip a machine already chosen" rule matters more than it looks. Without it, three replicas can land on one machine, and the replication factor becomes a lie.

What a read returns#

Three numbers control everything here:

  • N: how many copies of each key exist. Here, 3.
  • W: how many replicas must confirm a write before the client hears "done".
  • R: how many replicas a read asks before answering.

A small worked example. N = 3, W = 2, R = 2. A write reaches replicas A and B and is confirmed. Replica C has not got it yet. A read now asks any two replicas. If it asks A and C, A has the new value. If it asks B and C, B has it. If it asks A and B, both do. There is no way to pick two out of three that misses both A and B, so when R + W is greater than N, every read overlaps the latest completed write. The read compares versions and returns the newest.

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.

Play with the dials above. 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 behaves like a cache: fast, always writable, often 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. It survives one node being down on either side and still overlaps.

"Stale" here is the difference between strong and eventual consistency, which strong vs eventual consistency shows live. The quorums section of the replication chapter goes deeper, and the CAP theorem chapter explains why staying writable during a network split means giving up some consistency.

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 order between them. That is a separate problem (the first deep dive), 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. The picture at the top showed one write through the middle of it. There is no leader in it, no config service in the request path, and no box every request has to pass through. That is the design goal, not minimalism: 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 arrows lead to 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. Each node keeps that copy up to date through gossip: every second or so, nodes swap "who is alive" news with a few random peers, and the news spreads through the cluster like a rumour. (Failure detection and gossip explains how.) 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 keeps the slow tail of response times down.
  5. Each replica appends the write to its own commit log and memtable. Nothing is updated in place. (The third deep dive explains why.)
  6. Repair runs beside all of this: hints for nodes that were absent, read repair on values a read already fetched, and Merkle trees for the keys nobody reads. The second deep dive explains all three.
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#

Two people share a shopping cart. At almost the same moment, one adds milk on their phone and the other adds eggs on a laptop. Both writes reach the store, through different coordinators. Which one is the cart now?

Two writes to the same key arrive at the same time. What does the store keep?

  1. First try
    Last write wins, by timestamp
    Stamp every write with the time on the coordinator. When two versions meet, keep the one with the later time and throw the other away. Simple, and many deployments use it.
    Where it breaks: One write is silently lost: the cart has milk or eggs, never both. Worse, clocks on different machines disagree by milliseconds or more, so the write that really came second can carry the earlier time and lose.
  2. Better
    A version number per key
    Each write bumps a counter: version 4 becomes version 5. A replica keeps the higher number.
    Where it breaks: Both writers read version 4 and both write version 5, on different coordinators. The store sees two different "version 5" values and has no way to know they were concurrent and not the same write.
  3. What you ship
    Version vectors, siblings, and the application merges
    A version vector is a small list of counters, one per node that has coordinated a write to this key, like "n1:4, n3:2". Comparing two vectors tells you whether one write came after the other or neither saw the other. If they are concurrent, keep both as siblings and return both on the next read. The application merges them, because only it knows a cart is a union and a counter is a sum.

Here is the third rung, one message at a time.

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.

Say out loud which one you are choosing and what it costs. Last write wins is defensible for a value that is simply overwritten, like a user's profile photo. It is not defensible silently.

Clocks are the root of this whole problem. Time and clocks shows how far apart machine clocks drift, and its section on vector clocks walks through the comparison step by step. If you want the merge to happen without application code, CRDTs are data types built so that any two copies can always be merged automatically.

Nodes are always missing#

At a few hundred nodes, some node is always restarting, upgrading, or briefly unreachable. The question is what happens to writes meant for it, and how it catches up afterwards.

A replica is down. What happens to writes for its keys?

  1. First try
    Fail the write until all replicas answer
    Set W = N. A write only succeeds when every replica has it, so no copy is ever behind.
    Where it breaks: With hundreds of nodes something is always down, so some keys always fail to write. The store is correct and unusable.
  2. Better
    Write to the replicas that are up, and move on
    With W = 2, the two live replicas accept the write and the client hears "done". The missing replica is ignored.
    Where it breaks: The down replica comes back with old data and nobody tells it. Every read that happens to ask it gets a stale answer, and the copies drift further apart over time.
  3. What you ship
    Hinted handoff, read repair, and Merkle tree anti entropy
    Writes for a down node are parked on a neighbour as hints and delivered when it returns. Reads compare the copies they fetched and fix stale ones on the spot. In the background, replicas compare Merkle trees to find and fix everything else, including keys nobody reads.

Switch between the three repair paths below.

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. The cost is a window where the hint holder carries data it does not own.

Anti entropy just means "background work that stops copies drifting apart". The tool it uses is a Merkle tree. Start with a hash: a short fingerprint of some data, where the same input always gives the same fingerprint and any change gives a different one (the hashing article shows one live). Hash every key and value. Then hash those hashes in pairs, then hash the pairs in pairs, until one hash is left at the top: the root. If any key changes, its fingerprint changes, and so does every hash on the way up to the root.

Two replicas compare roots. Same root, same data, finished. Different root, compare the two hashes one level down, and only follow the side that differs. Try it: make a key stale and watch which boxes get compared.

Two replicas, compared by Merkle tree
root a36b
bec1
06b3
f3e0
2303
83b1
e920
8510
51ae
1e60
68aa
663b
7f7a
e13d
dd3b
Replica 2's tree, built over its keys. Green boxes matched replica 1, red ones did not.
7 hash comparisons for 8 keys
The roots differ, so the walk goes down, but only into branches whose hashes differ. 7 comparisons found the 1 stale key. Branches in the faint colour were never looked at.
Click a key on replica 2 to make it stale (v5 instead of v7). With a million keys and one stale key the tree is about 20 levels deep, so the walk is about 40 comparisons instead of a million.

This is the answer to the question at the top of the page. A node returning after an hour does not compare every key it owns. It compares a root hash with its peers and goes down only into the parts of the tree that disagree.

The storage engine#

Now inside one replica. A node may hold 8TB of keys, takes a steady stream of writes, and can lose power at any moment. How does it store them?

How does one node store its keys on disk?

  1. First try
    A hash map in memory
    Keep every key in a hash map in RAM. Reads and writes are instant.
    Where it breaks: A crash or restart loses everything, and 8TB does not fit in memory anyway.
  2. Better
    A hash map in memory plus a commit log
    Before changing the map, append the write to a log file on disk. Appending to the end of a file is the fastest thing a disk does. After a crash, replay the log to rebuild the map.
    Where it breaks: The data still has to fit in memory, and the log grows for ever, so a restart replays hours of writes before the node can serve a single read.
  3. What you ship
    An LSM tree: commit log, memtable, sorted files
    Writes go to the commit log and to a sorted in memory table (the memtable). When the memtable fills, it is written to disk as an immutable sorted file and the log is trimmed. Reads check the memtable, then the files newest first, with a Bloom filter per file to skip most of them. Files are merged in the background (compaction).

LSM tree stands for log structured merge tree. The reason to pick it over a B-tree (the structure most relational databases use, which updates pages in place) is that every write is an append. There are no random disk writes and no page splits on the write path, and immutable files are trivially safe to copy for repair or to stream to a new node. Storage engines compares the two families in depth.

A Bloom filter answers "is this key in that file?" with either "definitely not" or "maybe", using only a few bits per key. Most files say "definitely not", so a read opens only one or two. The web crawler page has a 16 bit Bloom filter you can play with.

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 to disk 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 uses disk and CPU in the background, and a node 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#

Push the dial and watch what the cluster does at each level of damage.

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 and not 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 and not the slowest, and nothing is coordinating anything.
  2. one node down · Healthy. A replica restarts. W = 2 is still met 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 the three replicas for some key range are gone. W = 2 cannot be met, so writes to those keys fail. That is correct, and worth saying out loud: the store refuses a write it cannot make durable.
  4. partition · Down. The network splits the cluster in two. 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 when the sides rejoin, as sibling versions the application has to merge.
  5. repaired · Healthy. The partition heals. Hints are delivered, read repair fixes the popular keys right away, and anti entropy walks the Merkle trees to catch everything nobody has read. Conflicts show up 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 merge, and there are no multi key transactions.Availability matters more than ordering and the data has a natural merge.
Version vectors over last write winsConcurrent writes are detected, not silently discarded.Applications must handle siblings, and the vector grows with the number of nodes that coordinate writes.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 uses 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 copies can drift, which is all of them.

The level bar#

One moment from this round, answered three ways. Pick the one you think is staff level before the labels appear. (What changes at senior and staff explains the gap.)

Pick the staff level answer

InterviewerA product team wants W = 1 and R = 1 because the store feels slow. What do you say?

Show which is which without guessing
  • Answer A: Mid level. Gives the guarantee away without saying what it was.
  • Answer B: Senior. The right arithmetic and the right cost. It stops there and leaves the team to decide blind.
  • Answer C: Staff. Turns a global setting into a choice per use, and checks the premise before trading correctness away.

Then run the question cold, with a 45 minute timer and no notes, and score the run on the five things this question is really testing.

Score your last run
0 / 10
  1. Placement
  2. Quorum arithmetic
  3. Concurrent writes
  4. Failures
  5. One node on disk
Pick 0, 1 or 2 on every row. Be the strict interviewer, not the kind friend.

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 its replicas are the next N distinct physical nodes after it, skipping extra 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 a small slice 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, not 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 go down and compare children, so traffic follows how much actually differs, not how much data exists. Read repair handles popular keys right away, because reads already contact R replicas, and anti entropy fixes the keys nobody ever reads.
Explains the cost model instead of just naming the structure, and ties the two repair paths together.

Checkpoint#

Checkpoint

1. How does a returning node find out which keys are stale without comparing them all?

2. N = 3, W = 2, R = 2, and two clients write the same key at the same time. What does the store do?

3. Why does a delete write a tombstone instead of 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 overlaps every completed write, so a read sees the latest one. 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, instead of 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 going down 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.