Designing a Key-Value Store

It is 8:15 pm on a Saturday, an IPL match is on, and Meera's family in Bengaluru is building one dinner order in a food delivery app. Meera adds idli from her phone; her brother adds vada from his. At that moment one of the machines holding carts loses power. Nobody notices. Both items are saved, the cart still loads, and when the machine returns it quietly catches up on what it missed. That takes a store that keeps every cart on several machines, stays writable and repairs itself.

Meera's cart, key cart:17, is this chapter's running example. A key-value store does one job: given a key, store a value or return it. On one machine this is a hash map; the hard part is billions of keys across hundreds of machines, some of them always failing. Written down, the requirements are:

A quick estimate shows why one machine will not do. Two billion keys at about 1 KB each is 2 TB; three copies make 6 TB. Add a peak of 300,000 requests per second and a rule that no single failure may stop writes, and one machine is out of the question. As Databases: Replication, Partitioning and Choosing a Store showed, we must partition the keys across machines and replicate every partition.

put(key, value, context)  ->  ok
get(key)                  ->  (values, context)

context carries version information for conflicts, which is also why get can return more than one value.

The first question is which machine holds cart:17. The databases chapter showed that placing keys by hash(key) mod the number of servers moves almost every key when a server is added or removed. Consistent hashing fixes this. Take the output range of a hash function, say 0 to 2^32 - 1, and bend it into a circle. Each server is hashed to a position on this ring, and each key is hashed onto the same ring. A key belongs to the first server clockwise from it.

When a server joins, it takes over only the keys between itself and the server before it. When a server leaves, only its keys move, to the next server clockwise. With 10 servers, one change moves about a tenth of the keys on average, instead of nearly all of them.

One spot per server is not enough. Arcs come out uneven by chance, and when a server leaves, its whole arc lands on one neighbour, which may then overload as well. Both problems shrink when every server holds many spots.

With virtual nodes, each physical server is placed on the ring many times, for example 100 times, by hashing names like A#1, A#2 and so on. Each server then owns many small arcs scattered around the ring instead of one large arc. The two layouts compare like this:

Virtual nodes win on every row except the size of the ring table, and even that stays small.

Every node and every client library keeps a copy of the ring, so any of them can work out where a key lives without asking a central server. How they all learn about changes to the ring is the job of gossip, later in this chapter.

Spreading keys is half the job; the other half is keeping copies. Each key is stored on N nodes, usually 3. The first server clockwise from the key is its coordinator, and the key is also copied to the next N − 1 servers clockwise. This list of N servers is the key's preference list. While walking the ring, skip virtual nodes of a server already on the list, so the N copies sit on N different machines, ideally in different racks or zones.

Each request then picks how many replicas must answer. A write succeeds when W replicas confirm it; a read waits for R replies and returns the newest version. As in the databases chapter, R + W > N means every read overlaps the last successful write. Three common settings show the trade:

Latency follows from the numbers. A write with W = 2 waits for the second-fastest of three replicas, so one slow or dead replica does not slow it down. Raising W or R buys freshness and costs speed and availability; lowering them does the opposite. That trade, made per request, is tunable consistency. Even the strictest setting is not the linearizable model from Abstractions That Hide the Network, as the box below explains.

Put the pieces together and follow a single read of cart:17. The client library hashes the key, finds the coordinator on its own copy of the ring, and sends the request straight there. A simpler client can send to any node, which forwards the request; that costs one extra hop.

A put takes the same route: the coordinator stamps a new version, writes it locally, sends it to the other replicas and replies once W of them confirm. Neither path goes through a master or a central directory, which is why the store keeps working when any single machine fails. Now size the cluster that holds every cart:

Here traffic, not storage, sets the size of the cluster. Always work out both, because either one can be the limit: a store of large documents read rarely would flip it.

Quorums decide how many replicas take a write, but not what happens when two writes to one key overlap. Without a leader, two clients can update the same key at the same moment through different nodes. Back to the opening: Meera adds idli from her phone while her brother, whose app still shows the older cart, adds vada. Both writes succeed on different replicas, and the store now holds two versions of cart:17.

The simplest answer is last write wins: stamp every write with a time and keep the highest. The careful answer is a vector clock, a small list of counters with one entry per node that has coordinated a write to this key. Version A descends from version B if every counter in A is at least as large as in B. If neither descends from the other, the writes were concurrent, and both are kept. Follow cart:17 through the race:

Vector clocks found the conflict, and the app, which knows what a cart means, resolved it by keeping every item.

Vector clocks are not the only answer, and for some data they are more machinery than the data deserves. The question is what a lost write costs. The two approaches compare like this:

Choose by what a lost write costs: a lost "last seen" time is harmless, while a lost dosa is a complaint from a hungry family.

Conflicts come from writes that overlap. The other source of drift is a replica that misses writes altogether, and the next page starts with short outages.

Nodes go down for a few minutes all the time: a restart, a deploy, a network blip. With a strict quorum, a write needs W of the key's N home replicas; if two of three are down, the write fails. For a cart, that is unacceptable.

A sloppy quorum lets the write land on the next healthy nodes on the ring instead. The stand-in stores the value with a hint naming the real owner. When the owner returns, the stand-in hands the write over and deletes its copy. This is called hinted handoff.

The store stayed writable, and the copy found its way home. Now imagine B had been down too: W = 2 would have been met by A and the stand-in D. A read that reached B and C after they came back, before the hints arrived, would miss the new value entirely. That is the price of a sloppy quorum: it trades some consistency for availability, and R + W > N stops being a guarantee while hints are outstanding.

Hints cover short outages. If a node is gone for hours, a hint is lost, or a disk is replaced, replicas drift apart: if node C was away for six hours and its hints expired, its copy of cart:17 and thousands of other keys are now stale. They need anti-entropy: a background process that compares two replicas and copies over whatever differs.

Comparing key by key would mean sending hundreds of millions of keys, or their hashes, across the network for every pair of replicas. A Merkle tree avoids that. Split the key range into buckets and hash each bucket's contents. Then hash pairs of those hashes, and pairs of those, up to a single root hash. Two replicas compare roots first; if the roots match, the replicas are identical and the check is over.

Each level halves the search, so when only a few buckets differ, the work grows with the depth of the tree, not its size. With about a million buckets, one difference is found in about 20 levels of comparison, and only the keys in differing buckets cross the network.

Hints and routing both assume every node knows which nodes exist and which are alive, without a central server that could itself fail. Nodes use a gossip protocol, the way news spreads in a hostel: you tell a random friend, who tells another, who tells another.

• Each node keeps a membership list: every node it knows of, with a heartbeat counter and the local time that counter last went up.
• Every second or so, a node raises its own counter, picks a random peer, and the two swap lists, keeping the higher counter for each node.
• If a node's counter has not risen for a while, say 10 seconds, it is suspected. If it stays silent longer, it is marked down and requests route around it.
• A new node contacts a few well-known seed nodes to join, so the cluster never splits into groups that do not know about each other.

Because the number of nodes that know a piece of news roughly doubles every round, it reaches a 1,000-node cluster in around 10 to 15 rounds: 10 to 15 seconds at one round a second. Each node contacts only one peer per round, so the number of messages each node sends stays flat as the cluster grows.

Everything so far is leaderless: any replica accepts writes, and disagreements are settled later. The other common design keeps the partitions but gives each one a leader. A partition's 3 replicas form a small consensus group, as in Consensus, Leader Election and Coordination; the leader puts every write to that partition in order and confirms it once 2 of the 3 replicas have it. The two designs differ on almost every axis:

Neither design wins everywhere; what the data can tolerate decides.

So far every replica sat in one data centre; large stores spread them over several, so a whole site can fail without losing data. Say 3 replicas sit in Mumbai and 3 in Delhi, so N = 6, and a round trip between the two sites is about 30 ms. A quorum must now say which replicas it counts.

Most traffic uses local quorums and copies to the other site in the background. The two sites then behave like the multi-leader setup in the databases chapter: both can update the same key at once, and versions or last write wins must settle it. Keep each user's writes in their home site where you can, so conflicts stay rare.

The whiteboard design hides the work of running it. Four realities decide whether a cluster stays healthy.

None of this shows on a design diagram, yet it decides whether the cluster survives its first big sale. The write path under Going further shows the files that compaction merges.

Optional: what happens inside one node when a put arrives, for readers who want to see the storage engine.

So far each node has been a box that stores keys. Inside, most stores of this kind use a log-structured design that turns scattered random writes into fast sequential ones.

A read checks the memtable first, then the SSTables from newest to oldest. To avoid opening every file, each SSTable has a Bloom filter, a compact bit array that answers either "definitely not here" or "possibly here". Most files are skipped after a quick check in memory.

For the prompt "design a key-value store", here is this chapter fitted to the 40 minutes of design time in a 45-minute round.

Pick conflicts if they ask about two people editing one cart; failure handling if a node dies for an hour or a disk is replaced; the leader-per-partition design if they add a wallet or compare-and-set.