Design a Key-Value Store

A key-value store does two things: put(key, value) and get(key). On one machine that is a hash map with a log. Spread it over thirty machines that fail, and you must decide where each key lives, how many copies to keep, what a read returns while copies disagree, and how they heal.

Interviewers ask it to test whether you are precise about guarantees; "it's eventually consistent" is not an answer. It follows A Framework for Any Design and leans on Consistent Hashing and Replication.

#Clarify the problem

The design swings on one answer: during a failure, do you refuse writes or accept them and sort out conflicts later? Ask it early.

• Functional: get, put and delete by key; the caller chooses how many replicas must answer.
• Non-functional: writes are accepted even when some replicas of a key are down; p99 under 10 ms within the region; at the default write level, an acknowledged write survives any single node failure; the cluster grows by adding nodes without downtime.
• Out of scope: range scans, secondary indexes, multi-key transactions, multiple regions.

#Estimate the scale

• Raw data: 2 billion keys × 10 KB = 2 × 10¹³ B = 20 TB.
• Write bandwidth: 50,000 × 10 KB = 500 MB/s from clients; with 3 copies, 1.5 GB/s of replica writes across the cluster.
• Read bandwidth: 150,000 × 10 KB = 1.5 GB/s back to clients.

#Define the API

get(key, read = "quorum")
  -> { values: [v1, ...], context }   one value normally; several after a conflict
  -> NOT_FOUND
  -> UNAVAILABLE                      fewer than R replicas answered in time
 
put(key, value, context?, write = "quorum")
  -> OK                               W replicas have it in their commit logs
  -> UNAVAILABLE                      fewer than W confirmed in time (it may
                                      still be stored on some replicas)
  -> TOO_LARGE                        key over 256 B or value over 1 MB
 
delete(key, context?, write = "quorum")   writes a tombstone
 
read / write levels: "one" | "quorum" | "all"   (R or W out of N = 3)

The decision worth arguing is the context. Without it, put overwrites and the newest timestamp wins, which silently drops one of two concurrent writes. With it, the store can tell "this replaces what you read" from "these happened in parallel" and hand both back to merge. It costs callers complexity, so make it optional: no context means last-write-wins.

Note the honest UNAVAILABLE on put: a write that reached 1 of 2 required replicas reports failure but may still be read later. Callers treat it as "unknown" and retry.

#Model the data

record   (on disk, sorted by key)
  key              up to 256 B
  value            up to 1 MB, about 10 KB on average
  version          version vector, e.g. {B: 3, C: 1}, about 16 B per entry
  timestamp        8 B, used only for last-write-wins and tombstone expiry
  tombstone        1 bit: this key was deleted
 
cluster metadata (same on every node, a few MB)
  ring             node -> its virtual-node positions on the hash ring
  membership       node -> zone, status (up / suspected / leaving), heartbeat

Every operation is a single-key lookup or overwrite. Writes are a quarter of the traffic, but that is 5,000 replica writes a second per node, every one to a random key. An LSM tree turns them into sequential appends; Indexing explains how. Its cost is reads that may check several files, which Bloom filters keep cheap. A B-tree would be the pick if writes were rare.

#High-level design

There is no leader. Any node can coordinate any request, and every node knows the ring. Keys sit on a hash ring with about 100 virtual nodes per server. A key's preference list is the first 3 distinct servers clockwise from its hash, chosen so the 3 copies land in 3 different zones.

#Deep dives

#Partitioning with consistent hashing

Each server gets about 100 points on the ring (virtual nodes); Consistent Hashing covers why. What a replicated store adds:

• Skip duplicates when walking the ring. Neighbouring points may belong to one server, so take the next 3 distinct servers in 3 distinct zones. Otherwise one zone outage can take every copy of a key.
• Rebuilds are parallel. A dead node's 2 TB is spread in small ranges over the other 29. If each streams its share at 100 MB/s, the rebuild takes 2 × 10¹² ÷ (29 × 10⁸) ≈ 690 s, about 11 minutes. From one peer it would take 5.6 hours.

#Replication and tunable quorums

Each key has N = 3 replicas. A write waits for W of them, a read for R. Replication traces why R + W > N makes read and write sets overlap. In the interview, pick values per use case:

Be precise about the limits. R + W > N guarantees overlap, not linearizability: concurrent writes both succeed, a failed write can still appear on one replica, and two reads during a write can see new then old. A sloppy quorum breaks even the overlap: if stand-ins like E met W, a read from the real owners can miss the write until the hint arrives. Say: "Quorum reads see the latest acknowledged write while the home replicas are reachable; during failures we choose availability and accept stale reads." More in CAP and Consistency.

#Resolving conflicts

Because the store takes writes during failures, two clients can update one key without seeing each other.

A version vector detects concurrency; it does not resolve it. Someone still merges, and a plain union has a flaw: an item removed on one device comes back from the other.

What would change my mind: if nearly every key has one writer, last-write-wins is simpler and safe enough.

#Detecting failures and repairing data

No central monitor decides who is dead. Every second, each node bumps its heartbeat counter and sends its membership table to a few random peers (gossip). If a node's counter has not moved for, say, 10 seconds, peers mark it suspected. That is an opinion: a slow node and a dead one look the same.

Treat temporary and permanent failures differently. A suspected node gets hinted handoff and no data moves; streaming 2 TB for a node that reboots in a minute does harm. Only when an operator, or automation with a long timeout, declares it gone do its ranges rebuild.

Three mechanisms repair data:

• Hinted handoff covers short outages. A node down for 10 minutes misses 5,000 writes a second × 600 s = 3 million writes, about 30 GB at 10 KB each, held by the stand-ins.
• Read repair fixes stale copies a read notices.
• Anti-entropy catches everything else, including keys nobody reads. Two replicas compare a Merkle tree, a tree of hashes over sub-ranges, and descend only into branches whose hashes differ. With 200 million keys per node and 2¹⁵ leaves, one wrong key costs a walk down 15 levels and a sync of one leaf of about 6,000 keys, not 200 million.

#The storage engine, in outline

A write is appended to the commit log, flushed to disk before the node confirms, and added to a sorted table in memory, later written out as an immutable sorted file. A read checks memory, then files from newest to oldest; a Bloom filter per file answers "definitely not here" without touching disk. Compaction merges files in the background.

Deletes are the trap. A delete writes a tombstone, which compaction may drop only after a grace period, typically days. If a replica was down when the delete happened and the tombstone is purged before repair reaches it, anti-entropy sees the old value on that replica, finds nothing on the others, and copies it back. The deleted key returns. So the grace period must be longer than the longest outage you repair from.

#Bottlenecks and failures

At 10x (600 TB replicated, 2 million operations a second), you need about 300 nodes. Consistent hashing handles the growth, but two things strain. Hot keys do not spread: a popular key still lives on 3 nodes. And compaction debt, where writes outrun merging, makes reads check more files; watch files per read and throttle writes early.

#What a strong answer sounds like

#Follow-up questions

#How would you run it in two regions?

Keep N replicas per region, acknowledge after a local quorum, and replicate to the other region asynchronously. Version vectors catch cross-region conflicts too; a region failure loses the last moments of unreplicated writes.

#How do you support range scans?

Hashing scatters neighbouring keys. Partition by key range instead, which keeps order but needs splitting of hot or large ranges. You cannot have both with one layout.

#How do you add expiry (TTL)?

Store an expiry time with the record; reads treat expired records as missing. Every replica expires it on its own, so no delete is sent, but the expired record must act as a tombstone through the same grace period. Drop it early and an older copy without the expiry, on a lagging replica, comes back through repair.

#How does a new node join without downtime?

It takes about 100 new points on the ring. For each range it now owns, the current owners stream it the data while new writes go to it as well. It serves reads only once streaming ends; then the old owners drop their extra copies.

#One key gets 100,000 reads a second. What now?

With R = 2, that is 200,000 replica reads on 3 nodes, about 67,000 each, over 4 times a node's normal 15,000. Cache it for a second in the coordinators, or read it with R = 1 if slight staleness is fine.

#What to practise next

A news feed stores its timelines in a store like this one. See how fan-out decides what gets written in Design a News Feed.