Design a Distributed Key-Value Store

Requirements

A key-value store exposes the simplest possible interface, put(key, value) and get(key), but doing it across hundreds of machines forces you to confront partitioning, replication, consistency and failure handling all at once. Amazon's Dynamo paper is the canonical blueprint and the basis for Cassandra, Riak and DynamoDB's ancestry.

Functional

• put(key, value) stores or overwrites a value; get(key) returns it; delete(key) removes it.
• Keys up to ~256 bytes, values up to ~1 MB (larger blobs belong in object storage).
• Optional per-key TTL.
• Per-request consistency level (for example ONE, QUORUM, ALL).

Non-functional

• Scale: tens of terabytes and hundreds of thousands of operations per second; grows by adding nodes.
• High availability: writes must succeed even when some replicas are down ("always writable").
• Low latency: p99 under ~10 ms within a region.
• Durability: an acknowledged write survives machine loss.
• Tunable consistency: callers choose between eventual and strong-ish reads.

Capacity Estimates

Assume 10 B keys, 1 KB average value, 300 K reads/s and 100 K writes/s at peak, replication factor 3.

Roughly 30 commodity nodes handle the load, so per-node performance is fine; the hard part is keeping 30 nodes (and later 300) correct while they fail independently.

API Design

PUT /v1/kv/{key}?consistency=QUORUM&ttl=3600
  Body:    <opaque bytes>
  Header:  If-Match: <version-context>      (optional, for read-modify-write)
  200:     { "version": "<context>" }

GET /v1/kv/{key}?consistency=ONE|QUORUM|ALL
  200:     { "value": "<base64>", "version": "<context>" }
  300:     { "siblings": [ ... ] }            (conflicting versions the client must merge)
  404:     not found

DELETE /v1/kv/{key}?consistency=QUORUM      (writes a tombstone)

The version context (a vector clock or version number) is opaque to the client: it reads it on get and passes it back on the next put so the store knows which version the write supersedes.

Data Model

Every node stores rows of the form:

Cluster metadata (which node owns which ring range, node status) is tiny and is spread by gossip or held in a small consensus store.

High-Level Design

flowchart LR
  Client --> LB[Load Balancer / smart client]
  LB --> C1[Coordinator node]
  C1 --> R1[(Replica A)]
  C1 --> R2[(Replica B)]
  C1 --> R3[(Replica C)]
  subgraph Node["Inside each storage node"]
    WAL[Commit log] --> MT[Memtable]
    MT -- "flush" --> SST[(SSTables + Bloom filters)]
  end
  R1 -. "gossip, hinted handoff, Merkle-tree repair" .- R2
  R2 -. "gossip" .- R3

There is no master. Any node can act as the coordinator for a request:

1. The coordinator hashes the key, finds its position on the consistent-hash ring and picks the first N healthy owners (the preference list).
2. Write: send to all N replicas, acknowledge the client after W replicas confirm.
3. Read: ask the replicas, return once R respond, resolving any version differences.
4. Background processes (gossip, hinted handoff, anti-entropy) repair whatever the fast path missed.

A "smart client" library that knows the ring can send requests straight to a replica and skip one hop.

Deep Dives

Partitioning with consistent hashing

Hashing keys with hash(key) % N reshuffles almost every key when N changes. Instead, place nodes and keys on a hash ring; each key belongs to the next node clockwise. Adding a node moves only the keys in one arc.

• Give each physical node many virtual nodes (tokens) so load is even and, when a node leaves, its ranges are spread across many survivors rather than dumped on one neighbour.
• More powerful machines get more virtual nodes, which handles heterogeneous hardware.
• See the Consistent Hashing article for the ring itself.

Replication and quorums

Each key is stored on N nodes: its owner plus the next N-1 distinct physical nodes clockwise (skipping virtual nodes of the same machine, and ideally spanning racks or availability zones).

With N replicas, a write waits for W acknowledgements and a read for R responses:

Handling failures: sloppy quorum and hinted handoff

If a replica in the preference list is unreachable, a strict quorum would reject the write. Dynamo instead uses a sloppy quorum: the coordinator writes to the next healthy node on the ring, tagging the data with a hint naming the intended owner. When the owner recovers, the stand-in hands the data back and deletes its copy.

• Failure detection: nodes gossip heartbeats every second; a node not heard from for a while is marked suspect locally. No central authority is needed.
• Permanent failures: anti-entropy repair compares replicas using Merkle trees per key range; only subtrees whose hashes differ are exchanged, so two mostly identical 1 TB replicas sync by sending kilobytes.
• Read repair: when a quorum read sees a stale replica, the coordinator writes the newest version back to it.

Versioning and conflict resolution

With always-writable replicas, two clients can update the same key on different sides of a partition. The store must detect and resolve the divergence.

• Last-write-wins (LWW): keep the value with the highest timestamp. Simple, used by Cassandra, but silently drops one of the concurrent writes and depends on clock sync.
• Vector clocks: each version carries {node: counter} pairs. If one clock dominates the other, the older version is discarded; if neither dominates, both are kept as siblings and returned to the client to merge (Dynamo's shopping cart merged by union).
• CRDTs: data types (counters, sets, maps) whose merge is mathematically guaranteed to converge; good when values have structure.

The storage engine on each node

Write-heavy key-value stores typically use a log-structured merge (LSM) tree:

1. Append the write to a commit log (WAL) on disk for durability.
2. Insert into an in-memory sorted memtable.
3. When the memtable is full, flush it as an immutable sorted SSTable file.
4. Reads check the memtable, then SSTables from newest to oldest; a per-SSTable Bloom filter skips files that cannot contain the key.
5. Compaction merges SSTables in the background, dropping overwritten values, expired TTLs and old tombstones.

Writes become sequential disk I/O, which is why LSM stores sustain very high write throughput. The cost is read and space amplification, which compaction strategy tuning trades off.

Scaling & Reliability

• Adding capacity: a new node takes over token ranges and streams only those ranges from existing replicas; clients keep working throughout.
• Rack and zone awareness: place the N replicas in different failure domains so a zone outage loses at most one copy.
• Multi-region: replicate asynchronously between regions and offer LOCAL_QUORUM so requests never wait on cross-region latency.
• Hot keys: a celebrity key can overwhelm its N replicas; add a cache in front, or split the value across sub-keys.
• Backpressure: bound coordinator queues and shed load rather than letting latency explode cluster-wide.
• Observability: track per-node p99, pending hints, compaction backlog and repair progress.

Trade-offs

Interview Tips

• Declare CP vs AP up front and justify it with the use case (shopping cart vs account balance).
• Build the design in layers: partitioning, then replication, then consistency, then failure handling, then storage engine.
• Explain the quorum condition R + W > N with a concrete N=3 example and be honest about its limits.
• Distinguish temporary failures (sloppy quorum, hinted handoff) from permanent ones (Merkle-tree anti-entropy).
• Expect a follow-up on concurrent writes; compare last-write-wins with vector clocks.
• Mention tombstones and compaction; they show you know how deletes really work.
• Use the capacity math to size the cluster and show per-node load is modest.

References

• System Design Primer: key-value store
• System Design Primer: CAP theorem and consistency patterns
• Dynamo: Amazon's Highly Available Key-value Store (SOSP 2007)
• Apache Cassandra documentation: Dynamo-style architecture
• Karan Pratap Singh: System Design, consistent hashing