Design a Key-Value Store

A key-value store is a simple kind of database. It just maps a key (a unique name) to a value (some data). The system does not care what the value is — it just stores it and hands it back. This is the model behind Amazon Dynamo, Cassandra, Redis, and Memcached. The whole API is only two calls, which is what makes this problem sneaky: it looks easy, but all the real work is turning a single-machine lookup table into a system that keeps working even when machines die.

Step 1: Requirements and rough sizing

The interface is tiny. There are only two operations:

put(key, value)   // save a value under this key
get(key)          // fetch the value saved under this key

Before designing anything, agree on what the system must do. These are the standard goals from the classic Dynamo-style problem:

• Small pairs — each key-value pair is under 10 KB (small).
• Big data — must hold far more data than one machine can fit.
• High availability — keeps answering quickly, even when parts of it break.
• High scalability — can grow to handle very large amounts of data.
• Automatic scaling — machines (nodes) are added or removed on their own as traffic changes.
• Tunable consistency and low latency — we can dial how fresh the data must be, and answers come back fast.

Why one server is not enough

A single-server store is easy: just keep a lookup table in memory. You can stretch it a bit by compressing data and by keeping only the hot (frequently used) data in memory while the rest sits on disk. But one box still fills up fast. Once your data is bigger than one machine can hold, you need a distributed key-value store — many machines working together. This is also called a distributed hash table.

Sizing the partitions

Pair size            ≤ 10 KB
Data set             >> 1 server's capacity   → must split across machines
Replication factor   N = 3                     → 3 copies = 3× raw storage
Merkle bucketing     1,000,000 buckets / 1,000,000,000 keys
                     → 1,000 keys per bucket   (used for repair, explained later)

These numbers explain the rest of the design. The data will not fit on one node, so we split it across many using consistent hashing. And we keep 3 copies of every key, so we still have the data if a node is lost.

Step 2: Splitting data with consistent hashing

The first job is spreading data across many servers. Two rules matter: spread it evenly, and move as little data as possible when a node is added or removed. Consistent hashing does both.

• Place the servers on a hash ring (imagine a clock face that the servers sit on).
• Hash each key onto the same ring. The key belongs to the first server clockwise from where it lands.

This gives you two nice properties for free:

Step 3: Replication for availability

To survive failures, save each key on N servers (you choose N). After finding the key's spot on the ring, go clockwise and pick the first N different servers. With virtual nodes you skip duplicates, so you really land on N separate physical machines.

N = 3   →   key0 is stored on the next 3 different nodes clockwise

To survive a bigger disaster (a whole data center losing power or network), put those N copies in different data centers connected by fast links.

Step 4: Tunable consistency with quorums

With N copies, reads and writes use a quorum — a minimum number that must agree. Define:

• N — how many copies (replicas) exist.
• W — write quorum: a write counts as done once W copies confirm it.
• R — read quorum: a read waits for R copies to reply.

A coordinator sits between the client and the nodes and does this work. Note: W = 1 does not mean only one copy exists — it means the coordinator only waits for one confirmation before saying "done."

The W, R, and N dials trade speed against freshness:

Here is why W + R > N gives strong consistency: if the write copies and the read copies add up to more than N, they must share at least one node. That shared node has the newest write, so any read sees the latest value.

Consistency models

• Strong — every read returns the newest write. Never stale (out of date).
• Weak — a read might miss a recent write.
• Eventual — a kind of weak consistency: given enough time, all copies catch up and match.

Dynamo and Cassandra use eventual consistency, and so do we. Strong consistency forces every copy to wait for all the others to agree before answering, which kills availability — exactly what we are trying to protect.

Step 5: Resolving conflicts with vector clocks

Eventual consistency lets two copies accept conflicting writes. Imagine get("name") returns the same value to two servers. One sets it to johnSanFrancisco, the other to johnNewYork. Now there are two versions, v1 and v2, and neither is clearly the winner.

For example, D([s0, 1], [s1, 1]) is an ancestor of D([s0, 1], [s1, 2]), so there is no conflict. But D([s0, 1], [s1, 2]) and D([s0, 2], [s1, 1]) are siblings — a genuine conflict, because each one is ahead in a different counter.

Two downsides worth naming: vector clocks push the conflict-resolving work onto the client, and the [server, version] list can grow forever. The fix is to cap the list length and drop the oldest pairs. Amazon says they never actually hit this limit in production, so it is an acceptable trade-off.

Step 6: Handling failures

Detecting failure — gossip

One server saying another is down is not enough — you want two independent sources to agree. Having every node check every other node works but does not scale. So use a decentralized gossip protocol:

• Each node keeps a membership list: who is in the cluster, plus a heartbeat counter for each.
• Each node bumps its own heartbeat now and then and shares heartbeats with a few random nodes, who pass them along.
• If a node's heartbeat stops going up for too long, it is marked offline — and that fact spreads ("gossips") outward until enough nodes agree.

Surviving failure — sloppy quorum + hinted handoff

A strict quorum gets stuck when some copies are down. So use a sloppy quorum instead: pick the first W healthy nodes for writes and the first R healthy nodes for reads, just skipping the offline ones.

For permanent failures, run an anti-entropy protocol using a Merkle tree. The idea: split the key space into buckets, hash each bucket, and build a tree of hashes from the bottom up. To compare two copies, walk down from the root. If the top hashes match, the data is identical. If not, you only dig into the branches that differ. The payoff: the amount of data you move is proportional to how much actually differs, not the whole dataset. With ~1,000 keys per bucket, one stale key syncs almost nothing.

Step 7: Architecture, write path, read path

The architecture is fully decentralized — every node has the same job, so there is no single point of failure (no one box that takes everything down if it dies):

• Clients call get(key) and put(key, value).
• A coordinator node sits between the client and the ring.
• Nodes are spread on the ring by consistent hashing; adding or removing nodes happens automatically.
• Data is copied across N nodes.

Inside each node, the write and read paths follow Cassandra's storage engine:

Write path:

1. Append the write to a commit log (so it survives a crash)
2. Update the in-memory cache (the "memtable")
3. When the memtable fills up, flush it to an SSTable on disk
   (an SSTable is a sorted list of <key, value> pairs)

Read path:

1. Check the in-memory cache → if it is there, return it
2. Not there → ask a bloom filter which SSTable(s)
   might hold the key
3. Read those SSTables, merge the results, return the value

The bloom filter is what keeps reads fast on a miss. Instead of scanning every SSTable on disk, it tells you (with a good guess) which files are even worth opening.

Wrap-up

The whole design is one idea, applied everywhere: decentralize everything. No coordinator is special, no node is a single point of failure, and consistency is a dial (W + R > N) rather than a fixed setting — which is exactly why Dynamo-style stores can promise both huge scale and "always-on" availability.

Practice

Check that the put/get flow and the failure tools stuck.