Key-value store

Design a Dynamo-style store: partition keys over a consistent-hash ring, replicate each key N times and let R and W quorums trade consistency for latency and availability.

Open reference design Interactive lesson Practice an interview

Arrows show permitted calls. Open the canvas to inspect route programs.

Component labels and connections

Clients (client) · calls: Load balancer
Load balancer (loadBalancer) · calls: Coordinator 1, Coordinator 2
Coordinator 1 (service) · calls: Storage ring (6 nodes)
Coordinator 2 (service) · calls: Storage ring (6 nodes)
Storage ring (6 nodes) (kvStore) · calls: none

Requirements

Functional

put(key, value) and get(key) for values up to ~1 MB (typically ~1 KB).
delete(key).
Optional: conditional put on a version for read-modify-write.

Non-functional

Single-digit-millisecond p99 for get and put inside a region.
Writes always accepted while a quorum of replicas is reachable (high write availability).
Tunable consistency per request; read-your-writes when R + W > N.
Scale horizontally to tens of TB and ~1M ops/s by adding nodes, with minimal data movement.

Worked estimates

Data
≈ 10 TB, ≈ 30 TB with N = 3
10B keys × ~1 KB
Nodes for storage
≈ 15 nodes
30 TB ÷ ~2 TB usable per node (leave room for compaction)
Nodes for throughput
≈ 23 nodes → plan ~30 for headroom and node loss
1M ops/s × (70% reads × R=2 + 30% writes × N=3 replicas written) ÷ ~100k replica ops/s per node
Data moved when a node joins
≈ 1 TB of 30 TB for the 31st node, vs nearly all of it with hash mod N
≈ 1/(nodes) of the keys with consistent hashing
Quorum latency
p99 is set by the 2nd replica, not the slowest — a reason not to pick W = N
W = 2 of 3 waits for the 2nd-fastest replica

API

GET /kv/{key}?consistency=quorum|one

Request: —

Response: 200 { value, version } · 404

PUT /kv/{key}

Request: { value } + If-Match: version (optional)

Response: 200 { version } · 409 version mismatch

A plain put is idempotent, so the coordinator can retry it safely.

DELETE /kv/{key}

Request: —

Response: 200

Writes a tombstone, garbage-collected after gc_grace (here 10 days) so a lagging replica cannot resurrect the value.

Data model

items — store: kv

Each key hashes to a token on the ring; its N = 3 replicas are the next 3 distinct nodes clockwise (the preference list).

Each node owns many virtual nodes (64 here, 256 in Cassandra's default), which spreads load evenly and spreads a failed node's load across all survivors.

Values carry a version (vector clock in Dynamo, timestamp in Cassandra) used to pick a winner and to repair replicas.

High-level design

Trace permitted calls in the architecture above, then open the reference canvas to inspect routing.

Deep dives

Quorums: why R + W > N

With N replicas, a write waits for W acknowledgements and a read for R replies. If R + W > N the read set and the last write set must overlap in at least one replica, so the read sees the latest acknowledged version (after picking the highest version among the replies). N = 3, R = W = 2 tolerates one slow or dead replica on both paths. R = 1, W = 3 makes reads fast and writes fragile; R = W = 1 is fastest but only eventually consistent. Note that quorum overlap is not linearizability: concurrent writes and sloppy quorums still need versions and conflict resolution.

Sloppy quorums, hinted handoff, read repair, anti-entropy

When a preference-list replica is down, Dynamo writes to the next healthy node with a hint and hands the data back when the owner recovers — writes stay available. Read repair updates stale replicas that a read happened to contact. Hints and read repair only fix keys that are touched, so a background anti-entropy process compares Merkle trees of key ranges between replicas and streams the differences.

Conflicts and deletes

Concurrent writes to the same key on different replicas produce siblings. Dynamo returns them with vector clocks for the application to merge (the shopping cart unions them); Cassandra uses last-write-wins timestamps, which silently drops one write if clocks skew. Deletes write tombstones that must outlive every replica's chance to sync; garbage-collecting them too early lets an old replica resurrect the deleted value.

Storage engine

Each node appends writes to a commit log and an in-memory memtable, flushed to immutable sorted files (an LSM tree) and compacted in the background. Writes are sequential and fast; reads may check several files, so bloom filters skip files that cannot contain the key. Compaction is the main source of latency spikes, as in the slow-ring scenario.

Failure scenarios

Packet loss between a coordinator and the ring

Scenario and expected result

From t = 1.2 s to 2.4 s, 5% of packets from coordinator 1 to the storage nodes are lost.

Expected: Lost requests time out after 30 ms and the coordinator retries with backoff. Because put is an idempotent overwrite and get has no side effects, retries are safe and nearly every request succeeds; the retries show up as a fatter tail.

Open this scenario

The storage nodes slow down 6×

Scenario and expected result

From t = 1.5 s to 2.3 s every storage operation takes 6× longer — compaction or a noisy neighbour on shared disks.

Expected: Quorum operations take 6–9 ms instead of 1–2 ms, and the p99 roughly quadruples. Everything still finishes inside the 200 ms deadline, so availability holds — the ring has spare workers. With less headroom the same slowdown would queue requests; ring utilisation is the signal to add nodes before the next compaction storm.

Open this scenario

Coordinator 2 loses its network path to the ring

Scenario and expected result

From t = 1.5 s to 2.0 s, coordinator 2 cannot reach any storage node, but it still answers the load balancer's health checks.

Expected: The health check only asks 'is the process up?', so coordinator 2 stays in the pool — a gray failure. What saves most requests is the least-connections policy: requests stuck on coordinator 2 keep its connection count high, so new requests flow to coordinator 1. Only the requests caught on coordinator 2 time out three times and fail. Compare round robin, which keeps sending half the traffic into the hole.

Open this scenario

Alternative designs

Reference

Inspect the architecture above and run the same scenario.

Weak quorums: R = 1, W = 1

Lowest latency and highest availability — any single replica answers. But R + W = 2 ≤ N = 3, so a read can land on a replica the write never reached: acknowledged writes become invisible to later reads. Run the baseline against it and the engine reports stale reads.

Open this alternative

Reference

Inspect the architecture above and run the same scenario.

Round-robin load balancing in front of the coordinators

Round robin ignores how busy each coordinator is. When one coordinator silently loses its path to the ring, round robin keeps giving it half the requests until a health check notices — and a shallow health check never does.

Open this alternative

Reference

Inspect the architecture above and run the same scenario.

Leader per shard with consensus (Raft/Paxos)

Each key range has one leader replicated by Raft (etcd, TiKV, CockroachDB, Spanner). Reads and writes are linearizable and there are no conflicting versions to merge. The cost: writes wait for a majority and stall during leader election, and a partitioned minority cannot accept writes — the CP side of the trade-off Dynamo deliberately avoided for the shopping cart.

Reference

Inspect the architecture above and run the same scenario.

Hash mod N instead of a ring

Trivial to implement, but changing N remaps almost every key, so every resize is a full reshuffle. Consistent hashing (or a directory of ranges) moves only about 1/N of the keys.

Clarifying questions

Value size distribution and total data size?
Read/write ratio and peak throughput?
Is eventual consistency acceptable, or do some operations need read-your-writes or compare-and-set?
Single region or multi-region?

Common mistakes

hash(key) mod N partitioning.
Claiming 'strong consistency' with R + W ≤ N.
Ignoring tombstones and resurrection.
Sizing for storage only and forgetting replication's effect on throughput.
Shallow health checks that miss gray failures.

Self-assessment rubric

requirements / levels 1-4

1/4 get/put only.
2/4 Adds value size and scale.
3/4 States latency, availability and consistency requirements.
4/4 Makes consistency tunable and explains which side of CAP the use case needs.

estimates / levels 1-4

1/4 None.
2/4 Total data size.
3/4 Node count from storage and from throughput, including replication.
4/4 Data moved on resize and quorum latency effects.

api / levels 1-4

1/4 get/put.
2/4 Adds delete and status codes.
3/4 Versions and conditional put.
4/4 Per-request consistency levels and idempotent retries.

data model / levels 1-4

1/4 A hash map.
2/4 Hash partitioning.
3/4 Consistent-hash ring with replicas.
4/4 Virtual nodes, preference lists and versioning.

high level / levels 1-4

1/4 One server.
2/4 Sharded servers.
3/4 Coordinators + ring + replication.
4/4 Explains read and write paths with quorums and failure handling.

deep dive / levels 1-4

1/4 None.
2/4 Mentions replication.
3/4 Quorum math and hinted handoff.
4/4 Read repair, anti-entropy, conflict resolution and tombstones.

wrap-up / levels 1-4

1/4 Stops.
2/4 Mentions monitoring.
3/4 Names compaction and hot keys as risks.
4/4 Proposes deep health checks, capacity signals and rebalancing procedure.