system-design

Design a Distributed Key-Value Store (like DynamoDB / Cassandra)

Difficulty: Tier 3 (deep) Asked at: Amazon, Google, Meta, senior/staff loops Time budget: 45–60 min

This is the question that separates people who use distributed systems from people who understand them. You’re building the storage layer everything else in this repo depends on — so every hard concept shows up at once: partitioning, replication, consistency, quorums, conflict resolution, failure detection. It’s a synthesis exam. Don’t attempt it until you’re comfortable with Part 4.

Prerequisites: Consistent Hashing, Replication, Quorums, Conflict Resolution / CRDTs, CAP

1. Requirements

Functional:

get(key) → value; put(key, value); delete(key).
Values are opaque blobs (bytes). Keys are strings.
Large scale: billions of keys, terabytes–petabytes, spread over many nodes.

Non-functional:

Highly available — always writable, even during failures (the Dynamo philosophy). Favour A over C.
Scalable & elastic — add nodes to grow; no downtime.
Low latency — single-digit-ms reads/writes.
Durable — no data loss; survive node/disk/rack failures.
Tunable consistency — let the caller trade consistency for latency/availability.

Out of scope: transactions across keys, SQL queries, secondary indexes (mention they’re hard here).

2. Estimation

1 PB of data, replicated 3× → 3 PB raw. At ~10 TB/node → ~300 nodes.
1M ops/sec. Spread over 300 nodes → ~3–4K ops/sec/node — very doable.
The interesting numbers aren’t capacity; they’re replication factor (N), and the R/W quorum sizes — those govern consistency and availability (§4).

3. API

put(key, value, [context])   → success / version
get(key)                     → value(s) + version context
delete(key)                  → tombstone

Note get may return multiple versions if there was a conflict (see §6c) — the client (or a merge function) resolves them.

4. The core decisions

This design is a set of interlocking choices. Present them as a system, not a list.

4a. Partitioning — where does a key live?

Consistent hashing (chapter). Hash the key onto a ring; the key belongs to the next node clockwise. Virtual nodes (each physical node owns many ring positions) keep the load even and make adding/removing nodes cheap — only a small slice of keys moves. 🚨 This is the reason to use consistent hashing over hash % N: elasticity without a full reshuffle.

4b. Replication — how many copies?

Each key is stored on the N nodes following its position on the ring (the “preference list”). N = 3 is typical. Replicas span racks/availability zones so a rack failure doesn’t lose all copies. Writes go to all N; reads can consult a subset.

4c. Consistency — quorums (the knob)

🚨 The central mechanism. With replication factor N, require:

W = replicas that must acknowledge a write.
R = replicas that must respond to a read.

If R + W > N, reads and writes overlap on at least one node → you read the latest write (strong-ish consistency). Tune the knob:

W=N, R=1: fast reads, slow/less-available writes.
W=1, R=N: fast available writes, slower reads.
W=2, R=2, N=3: balanced, R+W>N → consistent. The common default.
R+W ≤ N: eventual consistency, max availability/latency.

This is CAP made tunable per-operation. (Quorums)

5. High-level design

flowchart TB
    Client --> Coord[Coordinator node<br/>any node can coordinate]
    Coord -->|"put to N replicas"| R1[Replica 1]
    Coord --> R2[Replica 2]
    Coord --> R3[Replica 3]
    Ring[Consistent hash ring<br/>+ virtual nodes] -.routes.-> Coord
    Gossip[Gossip protocol<br/>membership & failure detection] -.-> Coord

Decentralized, peer-to-peer (Dynamo/Cassandra style): every node is equal, any node can coordinate a request, routing it to the N replicas for the key. No single master → no single point of failure. Nodes learn about each other and detect failures via gossip.

6. Deep dives

6a. Writes with hinted handoff

A write goes to the N preference-list nodes. If one is temporarily down, a different healthy node accepts the write on its behalf with a hint (“this really belongs to node X”), and hands it off when X recovers. 🚨 Hinted handoff = you can always accept writes (high availability) even when a replica is down — a key Dynamo idea.

6b. Reads, read-repair, and anti-entropy

A read queries R replicas. If they disagree (some stale), return the newest and repair the stale ones in the background (read-repair). For replicas that stayed divergent (missed writes), a background anti-entropy process using Merkle trees efficiently finds and reconciles differences without comparing every key. (Replication)

6c. Conflict resolution — the hard part

With high availability and no single master, two clients can write the same key concurrently on different replicas → conflicting versions. How to reconcile?

Last-write-wins (LWW): attach a timestamp; newest wins. Simple, but loses data on true concurrency and depends on clock sync. Cassandra’s default.
Vector clocks: track causality; detect true concurrency vs one-happened-before-the-other. Concurrent versions are returned to the client to merge (Dynamo’s approach — a shopping cart merges by union).
CRDTs: data types that merge deterministically (counters, sets) with no conflict at all. 🚨 Conflict resolution is where AP systems earn their complexity — name the options and their trade-offs.

6d. Membership & failure detection

Nodes gossip their view of the cluster — who’s up, who’s down, ring positions — so membership converges without a central registry. Failure detection is done via heartbeats/gossip with a suspicion mechanism (a node is suspected before declared dead to avoid flapping). (Gossip)

6e. Storage engine on each node

Each node stores its keys locally in an LSM-tree (memtable + SSTables + compaction) — optimized for high write throughput, which a KV store needs. (B-tree vs LSM)

7. Bottlenecks & scaling further

Adding capacity → add nodes; virtual nodes rebalance only a slice of keys, online.
Hot key → even consistent hashing puts a single hot key on N nodes; extreme hotspots need caching in front or key-splitting. (Hot Keys)
Wide rows / large values → chunk them; keep values bounded.
Cross-key transactions → not natively supported; push to the application or a layer above (this is a deliberate limitation you should name).

8. Trade-off summary

Decision	Chosen	Alternative	Why
Architecture	Peer-to-peer, no master	Leader-based	No single point of failure; always writable
Partitioning	Consistent hashing + vnodes	hash % N	Elastic — add nodes without full reshuffle
Consistency	Tunable quorum (R+W>N)	Fixed strong / eventual	Caller picks the C/A/latency trade per op
Availability	Hinted handoff	Reject writes on failure	Accept writes even when replicas are down
Conflicts	Vector clocks / LWW / CRDT	Ignore (last write)	Reconcile concurrent writes explicitly
Storage engine	LSM-tree	B-tree	Write-optimized

9. Follow-up questions

Explain R + W > N in one breath.

If the number of replicas a write must reach (W) plus the number a read must reach (R) exceeds the total replicas (N), then the read set and write set must overlap on at least one node — so any read is guaranteed to see at least one replica that has the latest acknowledged write, giving you strong-ish consistency. If R + W ≤ N they might not overlap, so a read can miss the newest write → eventual consistency. Tuning R, W, and N is how you slide between consistency, latency, and availability per operation.

Why consistent hashing instead of hash(key) % N?

With `% N`, changing N (adding/removing a node) changes the modulus, so *almost every key* remaps to a different node — a massive reshuffle that's catastrophic at petabyte scale. Consistent hashing places nodes and keys on a ring so adding/removing a node only moves the keys between that node and its neighbor — a small fraction. Virtual nodes further even out the load and smooth the rebalance. Elasticity without mass data movement is the whole point.

What does hinted handoff buy you?

Availability for writes during transient failures. If a replica in the preference list is temporarily unreachable, instead of failing the write (or blocking), another healthy node accepts it and stores a hint noting the intended owner; when that owner recovers, the hint is replayed to it. So the system keeps accepting writes at full W even with a node down, then heals — you get durability and availability without waiting for the failed node.

How do you detect and reconcile replicas that drifted apart?

Two mechanisms. Read-repair fixes divergence lazily: on a read, if replicas disagree, the coordinator returns the newest version and writes it back to the stale replicas. Anti-entropy fixes it proactively in the background using Merkle trees — each replica builds a hash tree over its key ranges, and two replicas compare trees top-down, descending only into subtrees whose hashes differ, so they find the exact keys that diverged without scanning everything. Together they bound how long replicas can stay inconsistent.

When would you NOT choose an AP key-value store like this?

When you need strong consistency and multi-key transactions — e.g. a bank ledger where you must never read stale balances or lose a write, or anything requiring ACID across records. Then you want a CP system (single-leader with consensus like a relational DB, Spanner, or a Raft-backed store) that sacrifices some availability during partitions for correctness. This KV store optimizes for availability and scale and deliberately gives up transactions and strong consistency by default — the right tool only when those trade-offs fit.

10. What junior / mid / senior answers look like

Junior: describes a hash map sharded across servers. May not reach replication or consistency. Fine for a warm-up but this is a senior question.
Mid: consistent hashing, N replicas, quorum reads/writes, mentions eventual consistency. Solid grasp of the mechanics.
Senior/Staff: the full synthesis — tunable R/W/N with the R+W>N insight, hinted handoff, read-repair
- Merkle-tree anti-entropy, vector-clock vs LWW vs CRDT conflict resolution with their trade-offs, gossip membership, LSM storage engine — and knows the deliberate limitations (no cross-key transactions) and when a CP system would be the right call instead.