Design a key-value store

In 2 minutes

I would build a leaderless, Dynamo-style store. Nodes sit on a hash ring with virtual nodes, a key belongs to the first node clockwise, and it is copied to three distinct nodes in different zones. Any node can coordinate: it finds the replicas, sends a write to all three, and succeeds when two acknowledge, and a read asks two and returns the newest. Because W plus R is greater than N, reads overlap writes, though the system is eventually consistent. Our estimate needs about 40 nodes for 20 billion records. When two clients write the same key concurrently, vector clocks detect it and keep both versions for the client to merge, while last-write-wins is available for simple cases. A down replica gets its writes through hinted handoff, and Merkle trees and read repair fix whatever drift remains. Nodes learn who is alive by gossip, with no master. On each node I would use an LSM engine: append to a commit log, update a memtable, flush to immutable SSTables with Bloom filters, and compact in the background. The cost is eventual consistency, which I would state plainly.

The question

Clarifying questions

• Do we favour availability or consistency when machines fail or the network splits? This is the heart of the design. Choosing availability leads to replicas that accept writes independently and repair later.
• How big are keys and values, and how many records in total? They set the number of machines, and very large values would call for chunking or an object store instead.
• What is the read-to-write ratio, and what latency do we need? A read-heavy load favours engines and caches built for reads. A write-heavy one favours a log-structured design.
• Do we need range scans, or only lookups by exact key? Hashing keys spreads load evenly but destroys key order, so range scans need a different partitioning plan.
• Can concurrent writes to the same key be merged, or must one simply win? It decides between keeping several versions for the client to merge and a simple last-write-wins rule.

Requirements

Functional

• Put a value under a key, get it back, and delete a key.
• Spread data over many machines, and add or remove machines while running.
• Keep several copies of every key so a machine failure loses nothing.
• Let each request choose its consistency, by how many replicas must answer.

Non-functional

• Latency: p99 under 10 ms for reads and writes inside one region.
• Availability of 99.99%, so writes continue when one replica is down.
• No data loss when any single machine fails, and none when a whole zone fails.
• Scales out by adding machines, with no central bottleneck.

Back-of-the-envelope estimates

Assumptions for this exercise:

• These numbers are assumptions for the exercise, not facts about any real system.
• 20 billion records of about 1 KB each, replicated 3 times. Each node stores at most 2 TB of it.
• 300,000 operations a second at peak, 80% reads and 20% writes. A read asks 2 replicas and a write goes to all 3.
• We leave 30% headroom for growth, repairs and compaction.

The arithmetic:

• Data with replicas: 20B × 1 KB = 20 TB; × 3 copies → 60 TB
• Nodes needed: 60 TB ÷ 2 TB per node = 30; plus 30% headroom ≈ 39 → 40 nodes
• Replica operations per second: Writes: 60,000 × 3 = 180,000. Reads: 240,000 × 2 = 480,000 → 660,000 replica operations a second
• Load per node: 660,000 ÷ 40 nodes → ≈ 16,500 operations a second each
• Bloom filter for a 1-million-key file: 1% false positives needs about 9.6 bits per key: 1M × 9.6 bits ÷ 8 → ≈ 1.2 MB, small enough to keep in memory

High-level design

A Client sends each request to any node, which becomes the Coordinator for that request. The coordinator hashes the key and looks on the ring, kept in the Ring and gossip view, to find the three Replicas that own it. There is no master, so any node can do this.

For a write, the coordinator sends the value to all three replicas and replies once enough of them, the write quorum, have acknowledged. For a read, it asks enough replicas, the read quorum, and returns the newest value. When a replica is down, the coordinator keeps the write in the Hint store and delivers it later.

Each replica keeps its data in a log-structured engine: a Commit log for durability, an in-memory Memtable for recent writes, and sorted SSTables on disk for everything older.

How a request flows

Write with a quorum

1. The client sends put(key, value) to any node, which acts as coordinator for this request.
2. The coordinator hashes the key and finds that replicas 1, 2 and 3 own it.
3. It sends the write to all three replicas at once. Replica 1 stores it and acknowledges.
4. Replica 2 receives the same write.
5. Replica 2 first appends the write to its commit log, a fast sequential disk write.
6. Then it updates its in-memory memtable, so reads can see the value.
7. It acknowledges the coordinator. With replica 1's ack, that makes 2 of 3 replicas.
8. Two acks meet the write quorum, so the coordinator reports success. Replica 3's ack, if it comes, is a bonus.

Read with a quorum

1. The client sends get(key) to any node.
2. The coordinator asks replica 2, and one other replica, for the key.
3. Replica 2 looks in its memtable first, which holds the newest data.
4. If the key is not there, Bloom filters rule out files that cannot hold it, and the rest are searched.
5. A file returns the value along with its version.
6. The engine passes the value up to the replica.
7. The replica returns the value and version to the coordinator.
8. The coordinator compares the versions from both replicas, returns the newest, and repairs a stale one in the background.

A replica is down

1. The client writes a key that replica 3 owns.
2. The write to replica 3 times out, because the machine is down.
3. The coordinator checks the gossip view, which also marks replica 3 as down.
4. It saves the write as a hint addressed to replica 3. Replicas 1 and 2 still give the quorum, so the client sees success.
5. Later, gossip shows replica 3 is back, and the hint store releases its saved writes.
6. The coordinator replays the saved writes to replica 3.
7. Replica 3 and replica 2 then exchange Merkle tree hashes of their key ranges to look for anything the hints missed.
8. Where hashes differ, replica 2 sends only the keys in that small range.

Deep dive: partitioning, replication and quorums

Place nodes and keys on a hash ring. A key belongs to the first node clockwise from its hash, and each node holds many small ranges, called virtual nodes, so load evens out and a new node takes a little from many neighbours. Adding or removing a node moves only about one key in N.

For replication, store each key on the next N distinct nodes clockwise, with N = 3. Skip nodes that share a rack or zone with one already chosen, so one failure cannot take out every copy. The list of nodes for a key is its preference list. Any node can act as coordinator, since each node holds the ring.

Quorums let each request trade speed against consistency. Let W be the number of replicas that must acknowledge a write, and R the number asked for a read. If W + R > N, then every read set overlaps every write set in at least one replica, so a read sees the latest acknowledged write.

• N = 3, W = 2, R = 2: balanced, and survives one failed replica for both reads and writes.
• W = 1, R = 3: very fast writes, slower reads.
• W = 3, R = 1: fast reads, and writes fail if any replica is down.
• W = 1, R = 1: fastest, but a read can return stale data.

Overlap is not a full guarantee. Writes that are still in flight, a replica that missed a write while down, or hinted writes counted toward the quorum can all let a stale read through for a short time. This is eventual consistency: replicas converge, but at any moment they may differ. Be clear about that in an interview, and say which parts of the system, if any, need something stronger.

Deep dive: conflicts, failure handling and repair

Two clients can write the same key through different coordinators at the same moment. Each replica accepts its write, so the copies now differ.

Last write wins attaches a timestamp to each write, and the higher one survives. It is simple, but clocks drift, so a newer write can be silently lost. Use it only when overwriting is acceptable.

Vector clocks record history. A version carries a small map from node to counter. When a node coordinates an update, it adds one to its own counter. Version A supersedes B if all of A's counters are at least B's. If neither supersedes the other, the writes were concurrent, and the store keeps both as siblings. The next read returns both, and the client merges them, for example by taking the union of two shopping carts, then writes the result back.

Hinted handoff. When a replica is down, another node accepts its writes, stores them with a hint naming the intended owner, and delivers them when the owner returns. Writes stay available, with a short window in which the real owner is behind.

Anti-entropy with Merkle trees. Hints cover short outages, but replicas can still drift. Each replica builds a tree over its key range. Leaves hash small groups of keys, and each parent hashes its children. Two replicas compare root hashes. If they match, the whole range is identical. If not, they walk down the mismatched branches and exchange only the few keys that differ. Reads also repair: when the coordinator sees a stale replica, it sends it the newer value.

Gossip. Each node, about once a second, shares what it knows about the other nodes with a random peer. Membership and health spread through the cluster in a few rounds without any central server. A node is suspected after missed heartbeats, and declared down once enough peers agree.

Deep dive: the storage engine: commit log, memtable and SSTables

A log-structured merge tree, an LSM tree, turns random writes into sequential ones. A write goes in two places:

1. The commit log. Append the write to a file on disk. Appending is fast and gives durability, because after a crash the node replays the log.
2. The memtable. Insert the write into an in-memory sorted structure, so recent data can be read without touching disk.

When the memtable reaches a size limit, say 64 MB, it is written out as an SSTable, a sorted, immutable file, and the log segment behind it is dropped. Files are never edited in place.

A read checks the memtable, then SSTables from newest to oldest, and stops at the first hit. Reading every file would be slow, so each SSTable has a Bloom filter that says either 'definitely not here' or 'maybe here'. For a million keys at a 1% false-positive rate, the filter is about 1.2 MB. A small index inside each file points to the right block, so checking one file costs about one disk read.

A delete cannot remove a key from old, immutable files. It writes a tombstone, a marker that hides older values. Tombstones are removed later, after a grace period long enough for every replica to hear of the delete.

As files pile up, reads must check more of them. Compaction merges several SSTables into one, keeping only the newest version of each key and dropping tombstones. Size-tiered compaction merges files of similar size: it is cheap for writes and uses more space. Leveled compaction keeps files in levels of sorted, non-overlapping ranges, each about ten times larger than the one above. It reads faster but rewrites data more often. Pick one by the workload.

Trade-offs

• Consistency model: Tunable quorums, with N = 3, W = 2, R = 2 over A single leader per range with consensus. There is no leader to fail, and requests stay available during partitions. The cost is eventual consistency and conflicts to resolve.
• Conflict handling: Vector clocks, with siblings merged by the client over Last write wins. No write is silently lost when two clients race. The cost is extra metadata and merge logic in the application.
• Storage engine: LSM tree over B-tree updated in place. Writes are sequential and fast, which suits a write-heavy load. The cost is reads that may check several files, and background compaction.
• Temporary failure: Hinted handoff over Fail writes while a replica is down. Writes stay available through short outages. The cost is a window where a read can miss the hinted data.

Common follow-up questions

What if some data needs strong consistency?
Use W = N or R = N for those requests, at the price of latency and availability. For true linearizability, run a consensus protocol such as Raft or Paxos per key range, with a leader. That is a different trade-off, because the range stops accepting writes while it elects a new leader.

How do you add a node to a running cluster?
The new node picks its tokens on the ring, learns the ranges it will own, and streams that data from the current owners while they keep serving. Once it has caught up, it takes over and the old owners delete the data. Because only some ranges move, the load on the cluster stays small.

How would you support range scans?
Hash partitioning scatters neighbouring keys, so a scan would visit every node. Instead, let a key have two parts: a partition key that is hashed, and a clustering key that keeps rows sorted inside the partition. Scans then work within one partition, which is what most applications need.

What do you do about a hot key?
One key lives on only three replicas, so a very popular key can overload them. Add a cache in front for reads, and for writes split the key into several sub-keys that are combined on read. Watch per-key traffic so you find these early.