Distributed Key-Value Store
Designing the database itself — consistent hashing, quorum reads and writes, and resolving conflicts when two nodes disagree
This design is worth working through because it assembles almost every distributed idea into one system: partitioning, replication, quorums, failure detection and conflict resolution.
1. Clarify and Scope
| Question | Assumption taken here |
|---|---|
| Interface? | get(key) and put(key, value) only |
| Value size? | Up to 1 MB |
| Consistency? | Tunable — the caller chooses |
| Scale? | Petabytes, hundreds of nodes |
| Availability? | Must accept writes during a partition |
That last row is the decision that shapes everything: this is an AP system by choice, so it stays writable when the network splits and resolves disagreements afterwards.
2. Put Numbers On It
1 petabyte of unique data
Replication factor N = 3
1 PB × 3 = 3 PB stored
NODES
commodity node ≈ 12 TB usable
3 PB ÷ 12 TB ≈ 250 nodes ← "hundreds", confirmed
KEYS
average value ≈ 1 KB
1 PB ÷ 1 KB = 1 trillion keys
TRAFFIC
1,000,000 ops/sec at peak
× 3 replicas touched per op = 3M internal messages/sec
÷ 250 nodes ≈ 12,000 messages/sec/nodeTwo numbers decide the whole design
250 nodes. If one machine fails twice a year, 250 of them fail more than once a day. Node failure is not an incident here — it is the normal state. A design that needs every replica present would never be up.
1 trillion keys. No table can map key → node at that size; the lookup table would itself need sharding. Placement must be computed from the key, by every node, with no shared state.
Those two facts force everything that follows: placement must be computed (section 4), and operations must complete while machines are missing (section 5).
3. Define the Interface
get(key, R) → { value, version } | not_found
put(key, value, version, W) → ack
delete(key, version, W) → ack (writes a tombstone, not a removal)
R, W how many replicas must answer — set per call, not per cluster
version what the caller last saw, handed back on the next writeThree operations. Notice what is absent.
No scan. No join. No begin_transaction. No secondary index.
That is not an oversight — every one of them would require coordination between nodes, and coordination is what this design gave up in order to stay writable during a partition. The interface is the promise; sections 4 to 8 are how it is kept.
A small interface is what buys the scale. When someone asks for range queries later, the honest answer is "that is a different system", and this step is why you can say so.
`R` and `W` are arguments, not configuration
Most stores fix consistency once, for everything they hold. Here the caller chooses per call.
A session token can be read with R=1 — fast, occasionally stale, and nobody is harmed. An account balance in the same cluster can be read with R=2 — slower, but it cannot miss a completed write.
Forcing one setting on all data means picking the strictest need and making everything else pay for it.
`version` travels in both directions
The store hands a version out on get and expects it back on put.
Without it, an arriving write is just bytes — the store cannot tell "an update to what I gave you" from "an update to something older that took a slow path". It would have to guess, and guessing loses data silently.
Carrying the version is what makes disagreement detectable, which is the entire subject of section 6.
4. Data Placement
With hundreds of nodes and constant membership changes, hash(key) % N is unusable — it moves most keys every time a node joins. The answer is consistent hashing with virtual nodes.
key → hash → position on the ring → first node clockwise
+ the next N-1 nodes for replicasReplicas are simply the next nodes on the ring
Once a key's position is known, its replicas are the next few distinct physical nodes clockwise. No separate placement table is needed, and every node can compute the location of any key independently.
That is what removes the need for a central coordinator on the data path — any node can receive any request and route it correctly.
5. Quorums
Data lives on N replicas. Two numbers control every read and write:
| Symbol | Meaning |
|---|---|
N | How many replicas hold each key |
W | How many must confirm a write |
R | How many must answer a read |
The one rule that matters
W + R > N → reads and writes overlap on at least one node
→ a read always sees the latest writeWith N=3, W=2, R=2: any two writers and any two readers must share at least one node, so a read cannot miss a completed write.
Set W + R ≤ N and reads may return stale data — sometimes a deliberate choice for speed.
Choosing the numbers
| Setting | Behaviour |
|---|---|
W=1, R=1 | Fastest, weakest — stale reads likely |
W=N, R=1 | Fast reads, slow writes, unavailable if any replica is down |
W=1, R=N | Fast writes, slow reads |
W=2, R=2, N=3 | The usual balance — survives one node down, strong enough |
W = N means no availability
Requiring every replica to confirm sounds safest. It means a single slow or dead replica blocks all writes.
With three replicas and 99.9% availability each, W=3 gives you 99.7% — worse than any individual node. Quorums exist precisely so that a minority can fail without stopping the system.
6. When Replicas Disagree
In an AP system, two clients can write the same key on opposite sides of a partition. Both writes succeed. Now what?
| Strategy | How | Cost |
|---|---|---|
| Last write wins | Highest timestamp survives | Simple; silently loses data |
| Vector clocks | Track causality; detect true conflicts | Correct; the client must resolve |
| CRDTs | Types that merge automatically | Elegant; only for certain data shapes |
Last write wins loses data, quietly
It relies on wall-clock timestamps across machines, which drift. Two concurrent writes mean one is discarded with no record and no error.
For a cache that is fine. For a shopping basket it means items vanish, and nobody can explain why.
Vector clocks instead record which node saw which version, so the system can tell "B came after A" from "A and B happened concurrently". Genuine conflicts are handed to the client, which knows the semantics — a basket, for example, can merge by taking the union of items.
7. Keeping Replicas In Step
Three mechanisms, at different timescales:
| Mechanism | When | Purpose |
|---|---|---|
| Read repair | During a read | A stale replica is corrected as a side effect |
| Hinted handoff | During a write | A neighbour holds writes for a node that is down |
| Anti-entropy | Background | Merkle trees find and fix divergence |
Merkle trees make comparison cheap
Comparing two replicas holding a billion keys sounds impossible. A Merkle tree hashes the data in a tree, so comparing roots is one comparison.
If the roots match, the replicas are identical — done. If not, descend only into the branches that differ. Finding a handful of differing keys among a billion costs a few dozen comparisons instead of a billion.
This is why anti-entropy repair is affordable enough to run continuously.
8. Membership and Failure
There is no coordinator, so nodes track each other by gossip — each periodically exchanges what it knows about the cluster with a few random peers. Information spreads exponentially and needs no central authority.
Failure detection is a suspicion level rather than a yes/no, because you cannot distinguish slow from dead. A node that misses heartbeats becomes increasingly suspect, and is only treated as down once several nodes agree.
9. What Was Traded Away
| Decision | Gained | Cost |
|---|---|---|
| Consistent hashing | Adding a node moves ~1/N of keys | Range queries are impossible |
| AP over CP | Writable during a partition | Reads can be stale; conflicts occur |
Quorum W=2, R=2 | Survives one node down | Two round trips per operation |
| Vector clocks | No silent data loss | Clients must resolve conflicts |
| Gossip membership | No coordinator to fail | Membership converges slowly |
| Key-value only | Simple, scalable | No joins, no transactions, no secondary indexes |
Follow-on Questions
Related Reading
- Partitioning and Consistent Hashing — the ring and virtual nodes
- Consistency and CAP — the AP choice made here
- Distributed Primitives — gossip, failure detection and consensus
- Storage Engines — LSM-trees, which stores like this use underneath