“Design a distributed key-value store” — the Dynamo question — is the deepest pure distributed-systems problem you are likely to get. It is less about boxes and arrows and more about the CAP trade-off, replication, quorums, and conflict resolution. A strong answer demonstrates that you understand why a system chooses availability over consistency and what that costs.
We will design a store like Amazon DynamoDB or Apache Cassandra: a get(key) / put(key, value) interface that stays available and fast under node failures and scales to petabytes.
Understanding the Problem
Functional Requirements
get(key)returns the value (or values, if conflicting) for a key.put(key, value)stores a value for a key.- That is it. A simple API is the point — the complexity is all internal.
Non-Functional Requirements
- High availability: Writes and reads succeed even during node failures and network partitions. “Always writable” is a design goal.
- Horizontal scalability: Add nodes to add capacity and throughput, linearly, with no downtime.
- Tunable consistency: Let the caller trade latency for freshness per operation.
- Low latency: p99 in the single-digit to low-tens of milliseconds.
- No single point of failure: Every node is equal — no master, no special coordinator.
Back-of-the-envelope Estimation
Assume an average of 100K writes/s and 200K reads/s, ~1 KB values plus a ~100 B key, a 3× daily peak, and that ~10% of writes create a new key (the rest overwrite).
- Peak load: 100K × 3 = 300K writes/s; 200K × 3 = 600K reads/s.
- Replica operations at peak (N = 3, R = 2): 300K × 3 + 600K × 2 = 2.1M ops/s. At ~20K ops/s per node, that is ~105 nodes. Far beyond one machine, so we partition.
- New data: 100K × 10% × 1.1 KB × 86,400 s ≈ 0.95 TB/day, or ~350 TB/year; × 3 replicas ≈ ~1 PB/year on disk.
- Storage nodes: 1 PB / ~10 TB per node ≈ ~100 nodes per year of data. Storage and throughput both land near 100 nodes, so we need cheap node additions (virtual nodes) and rebalancing that moves only ~1/M of keys (M = cluster size).
- Replication bandwidth at peak: 300K × 3 × 1.1 KB ≈ 1 GB/s cluster-wide, ~10 MB/s per node. Manageable, but repair traffic (hinted handoff, anti-entropy) must be throttled so it does not swamp it.
- Write-heavy workload: a sustained 300K writes/s is what points us at an LSM tree, with sequential appends instead of random in-place updates.
The CAP decision — state it explicitly
During a network partition you can have consistency or availability, not both. Dynamo-style stores choose availability and partition tolerance (AP): they keep accepting reads and writes on both sides of a partition and reconcile divergence afterward. This single decision drives everything below — replication, quorums, and conflict resolution all exist to make “always available, eventually consistent” workable.
Core Entities and APIs
API Design
get(key) -> value(s), context # context carries version info
put(key, value, context) # context enables conflict detectionThe context (an opaque version token / vector clock) is returned by get and passed back on put, so the system can tell whether a write is based on the latest version or an older one.
Node responsibilities
interface Node {
id: string;
tokens: number[]; // positions on the hash ring (virtual nodes)
store: LSMTree; // local persistent storage
peers: MembershipTable; // learned via gossip
}High-Level Design
Every node is identical and can receive any request. The node that receives a request acts as the coordinator for it: it locates the replicas responsible for the key, forwards the operation, and applies the quorum rules. There is no master.
The design rests on three pillars: partitioning (where does a key live?), replication (how many copies and where?), and consistency (how do we read and reconcile them?).
Deep Dive: Partitioning with Consistent Hashing
We cannot use hash(key) % M for a cluster of M nodes — adding or removing a node would remap almost every key. Instead, nodes and keys are placed on a hash ring, and a key belongs to the first node clockwise from its hash position.
def coordinator_for(key, ring):
pos = hash(key)
# First node clockwise from the key's position owns it
return ring.first_node_at_or_after(pos)Virtual nodes
If each physical node occupied a single ring position, load would be badly skewed and a node joining/leaving would dump its whole range onto one neighbor. So each physical node claims many positions (virtual nodes / tokens). This:
- Evens out load — keys spread uniformly across physical nodes.
- Spreads rebalancing — when a node joins, it takes small slices from many nodes rather than one; when a node leaves, its load disperses.
- Lets heterogeneous hardware carry proportional load (a bigger box gets more tokens).
When one of M nodes joins or leaves, only ~1/M of the keys move — the entire reason consistent hashing exists. (We use M for the cluster size and keep N for the replica count below.)
Deep Dive: Replication, Quorums, and Consistency
Replication to N successors
Each key is stored not just on its owning node but on the next N−1 nodes clockwise, giving N replicas (commonly N=3). The set of nodes responsible for a key is its preference list. Spreading the preference list across different physical machines (and racks/zones, skipping virtual-node duplicates) ensures copies survive failures.
Tunable quorums (W + R > N)
Rather than always writing all N copies synchronously, callers tune two knobs:
- W — replicas that must acknowledge a write before it is considered successful.
- R — replicas that must respond to a read.
The guarantee: if W + R > N, the read set and the write set must overlap, so a read always sees at least one replica carrying the latest write. One caveat: Dynamo-style stores use a sloppy quorum with hinted handoff, so during failures a write may be acknowledged by stand-in nodes outside the key’s preference list. In those windows W + R > N no longer guarantees overlap, and reads are only eventually consistent until the hints are delivered.
N = 3 (replicas)
Strong-ish: W = 2, R = 2 → 2 + 2 > 3 ✓ (overlap guaranteed)
Fast writes: W = 1, R = 3 → write returns after one ack
Fast reads: W = 3, R = 1 → read hits one node
Fast + lossy:W = 1, R = 1 → 1 + 1 < 3 ✗ (may read stale)This makes consistency a per-operation dial. Latency-sensitive, loss-tolerant workloads pick low W/R; correctness-sensitive ones pick W + R > N. The store does not decide for you.
Conflict resolution
Because writes can succeed on different replicas during a partition, two versions of a key can diverge. The store must reconcile them:
- Last-write-wins (LWW): attach a timestamp; the newest wins. Simple, but silently drops a concurrent write. Fine for caches and idempotent data.
- Vector clocks: each value carries a
(node → counter)version vector. On read, the system can tell whether one version descends from another (keep the newer) or whether they are concurrent (a genuine conflict). Concurrent versions are returned to the client to merge (as Dynamo does with shopping carts — union the items so nothing is lost).
Keeping replicas in sync
Three mechanisms repair divergence:
- Read-repair: when a read detects that a replica returned a stale value, the coordinator writes the newer value back to it — repair piggybacks on normal traffic.
- Hinted handoff: if a target replica is temporarily down, another node accepts the write and holds a “hint” to deliver it when the node recovers. This is what keeps the system always writable.
- Anti-entropy with Merkle trees: replicas periodically compare Merkle-tree hashes of their key ranges to find and fix divergence efficiently, without shipping all the data.
Deep Dive: Storage Engine and Membership
LSM-tree storage
Each node persists data with a log-structured merge tree — the standard engine for write-heavy stores (Cassandra, RocksDB, LevelDB):
- Writes go to an in-memory memtable plus a sequential commit log (durable, append-only → fast).
- When the memtable fills, it flushes to an immutable SSTable on disk.
- Background compaction merges SSTables and drops tombstoned (deleted) keys.
The win is that writes are sequential appends, not random in-place updates — ideal for the high write throughput an AP store must absorb. The cost is read amplification (a key may live in several SSTables), mitigated with Bloom filters per SSTable.
Membership via gossip
With no master, how does every node learn the ring and who is alive? Gossip: each node periodically exchanges membership and liveness state with a few random peers. Information about a join, a failure, or a token change spreads epidemically across the cluster in logarithmic rounds. Failure detection is similarly decentralized (e.g. phi-accrual), so there is no coordinator to lose.
Key Takeaways
-
Lead with the CAP decision. A Dynamo-style store chooses availability during partitions (AP) and reconciles later. Every other choice follows from that.
-
Consistent hashing with virtual nodes partitions the keyspace, evens out load, and moves only ~1/M of keys when one of M nodes joins or leaves.
-
Replicate to N successors and use tunable quorums.
W + R > Nforces the read and write sets to overlap, giving strong-enough reads; lower W/R trades freshness for latency, per operation. -
Design for conflicts, do not pretend they won’t happen. Vector clocks detect concurrent writes (merge them, as with a shopping cart); last-write-wins is the simpler, lossy alternative. Read-repair, hinted handoff, and Merkle-tree anti-entropy heal divergence.
-
LSM-tree storage + gossip membership. Sequential writes for throughput, and a masterless, epidemic membership protocol so there is no single point of failure.
This problem rewards depth over breadth: a candidate who can explain why W + R > N works, what a vector clock detects, and how hinted handoff keeps the system writable is demonstrating exactly the distributed-systems judgment senior interviews are looking for.
