Design a Key-Value Store

1. The problemCopy Anchor Link

A key-value (KV) store is a database that holds pairs: a unique key (e.g. "user:1042:cart") mapping to a value — an opaque blob (raw bytes; the database never looks inside it). It's the simplest kind of NoSQL database (non-relational: no tables-with-joins like SQL).

API — the only three operations:

  put(key, value)    write / overwrite
  get(key)           read
  delete(key)        remove

No joins, no secondary indexes (you can only look up by the key, never by the value), no multi-key transactions.

The trap: this is NOT the "design a cache" question.

  cache      a volatile (RAM-only, loseable) layer in FRONT of a database
             node dies → you lose speed, not data
  KV store   IS the system of record — the authoritative copy of the data
             node dies → you must NOT lose data

That one word — durability (a write, once acknowledged, survives crashes, disk failures, power loss) — drags in everything below: replication, quorums, conflict resolution, repair, and a real storage engine.

2. Requirements to clarify (first 5 min of the interview)Copy Anchor Link

Functional requirements = what the system does (the API above). Non-functional requirements = how well it does it — and these drive the whole design. Ask:

• Read-heavy or write-heavy? (ratio of gets to puts)
• Key/value size limits? (typical: ≤1 MB; rare huge values → chunk + manifest, §14)
• Latency target? (p99 = the time the slowest 1% of requests take; "single-digit ms p99" = 99% of requests finish in under ~10 ms)
• During a network failure: keep serving (availability) or refuse rather than serve stale data (consistency)? → decides §3

Back-of-envelope estimate (rough math to justify the architecture):

  100M keys × 10 KB avg   = 1 TB of raw data
  × 3 copies (replication) = 3 TB
  + say 100k requests/sec (QPS = queries per second)
  → too big for one machine's RAM, too much traffic for one machine's NIC
  → must DISTRIBUTE across many servers

Mention the single-server version first (a hash table in RAM; compress values; spill cold data to disk) — then break it on capacity. Shows you don't reach for distributed complexity by default.

3. CAP theorem — the framing decisionCopy Anchor Link

CAP stands for the three properties a distributed data system might want:

• C — Consistency: every read returns the most recent write (all servers agree on the current value). Note: this is a different "C" from the C in ACID; here it means "everyone sees the same, latest data."
• A — Availability: every request gets a (non-error) response, always — even if some servers are down.
• P — Partition tolerance: the system keeps working when a network partition happens — i.e. the network breaks such that some servers can't talk to other servers (cables cut, switch dies, datacenter link drops). Both halves are alive but can't see each other.

The theorem: when a partition happens, you cannot have both C and A. Proof by intuition: server group 1 gets a write, but the link to group 2 is down. Group 2 now either (a) answers reads with its old value → available but inconsistent, or (b) refuses to answer until the link heals → consistent but unavailable. There is no third option.

Partitions WILL happen in any real network, so P is not optional. The real design choice is: what do we sacrifice DURING a partition — C or A?

AP systems give eventual consistency: if writes stop, all copies converge to the same value eventually (seconds, usually), but a read during the window may see an older value.

Say the choice out loud in the interview. The canonical answer to this question is the AP / Dynamo-style design (Dynamo = Amazon's 2007 paper that invented most of this machinery). Everything below builds it. If the interviewer demands strong consistency instead: pivot to one leader (single boss server) per shard + a consensus protocol (Raft/Paxos — algorithms by which servers vote to agree on one value) — know that at a sentence level.

The whole design = four decisions:

1. where keys live → consistent hashing (§4)
2. how many copies must answer → quorum: N, R, W (§5)
3. who wins when two writes collide → vector clocks vs last-write-wins (§6)
4. per-request consistency dial → tunable R/W (§5)

4. Partitioning — which server has key X?Copy Anchor Link

Partitioning (= sharding): splitting the dataset across many servers because it doesn't fit on one. Done with consistent hashing: hash both keys and servers onto a circle (the ring); walking clockwise from a key's position, the first server you meet owns the key.

Full detail (why hash % N fails, arcs, adding/removing servers, virtual nodes, membership) lives in the separate consistent-hashing notes — one image as the anchor:

One-sentence version for THIS interview: "consistent hashing, so adding capacity moves only ~1/N of keys; virtual nodes (each physical server placed at many ring positions) for load balance; replicas = the next N−1 distinct physical machines clockwise, spread across racks."

5. Replication & quorums (the heart of the design)Copy Anchor Link

Replication: keeping each key on several servers (replicas) so no single failure loses data. Each key lives on N replicas (typically N=3): the ring owner + the next N−1 distinct nodes clockwise.

This design is leaderless: no single boss server per key — any replica can accept a read or write. (Contrast: leader-based replication, where one node takes all writes and followers copy it.) The node that happens to receive a client's request acts as the coordinator: it hashes the key, finds the N replicas, forwards the request to them, and collects answers.

The two knobs:

• W (write quorum) = how many of the N replicas must ack (acknowledge: "I durably stored it") before the coordinator tells the client "success"
• R (read quorum) = how many replicas must answer a read before the coordinator replies with the newest of their answers

The magic rule:

  R + W > N   →  the set of nodes that answered the read and the set that
                 stored the write MUST overlap on ≥1 node (pigeonhole)
              →  every read touches at least one up-to-date copy

Say the trade-off sentence: "R+W > N buys freshness but reduces availability (more nodes must be up); R+W ≤ N buys speed but allows stale reads." Choosing R and W per request = tunable consistency.

6. Conflicts & versioningCopy Anchor Link

Because the system is leaderless AND always-writable, two servers can accept concurrent writes to the same key (neither knew about the other — e.g. during a partition, or two clients racing). Now the key has two versions. Someone must decide the winner:

How to compare two vector clocks:

  X = [(A,2),(B,1)]     Y = [(A,3),(B,1)]
  every counter in X ≤ its counter in Y  →  X is an ANCESTOR of Y
                                            (Y already includes X) → drop X
  neither dominates the other            →  CONCURRENT → real conflict →
                                            keep both "siblings", let the
                                            reader merge (Dynamo's shopping
                                            cart: union of both carts)

Senior move: pick per use case, out loud. Cart / user data → vector clocks or merge-friendly data. Metrics, "latest status" → LWW is fine, losing one overwrite is harmless.

7. Failure handling — detect, bridge, repairCopy Anchor Link

Three beats, in this order. (A node = one server in the cluster.)

7.1 Detect — gossip protocolCopy Anchor Link

Problem: how does anyone know a node died? A central health-checker would be a SPOF (single point of failure) and a bottleneck. Instead, gossip:

• every node keeps a membership table: for each peer, a heartbeat counter (a number the peer increments every second to prove it's alive) and when we last saw it increase
• every round (~1 s), each node increments its own counter and sends its whole table to a few random peers
• receivers merge: for each entry keep the higher counter
• a node whose counter hasn't risen past a timeout → marked down (locally)

• News reaches all n nodes in ~log(n) rounds; each node sends O(1) msgs/round → scales with no central registry
• "Down" is a local, probabilistic opinion, not a global fact. Cassandra refines the fixed timeout with a phi-accrual detector (adapts the threshold to observed network jitter → fewer false alarms)

7.2 Bridge (temporary failure) — sloppy quorum + hinted handoffCopy Anchor Link

A strict quorum requires W acks from the key's N home replicas — so if one is down, writes fail. Availability lost. Instead:

• Sloppy quorum: accept acks from the first N healthy nodes walking the ring — a substitute node may stand in for a down home replica
• Hinted handoff: the substitute stores the value plus a hint — a note saying "this really belongs to node C" — and when C recovers, the substitute delivers the data to C and deletes its own copy

Risk to name: if the substitute ALSO dies before handoff, the hint (and possibly the write) is lost. And during a partition, sloppy quorums on the two sides can both "succeed" → more conflicts for §6 to resolve.

7.3 Repair (long-term divergence) — read repair + Merkle-tree anti-entropyCopy Anchor Link

Replicas drift apart (missed writes, lost hints). Two fixes:

• Read repair: a read already touches R replicas; if one returned a stale version, the coordinator pushes the newest version back to it. Free-ish, but only heals keys that get read.
• Anti-entropy (background sweep for everything else, incl. cold keys): each replica builds a Merkle tree = a tree of hashes over its key ranges — leaves hash small ranges, each parent hashes its children, the root summarizes the whole dataset in one hash. Two replicas compare top-down: roots equal → identical, ZERO data moved; roots differ → compare children, descend only into differing branches → ship only the divergent range.

  cost of comparison  ~ log(#ranges) hash exchanges
  cost of transfer    ~ amount of DIVERGENCE, not dataset size

8. Storage engine — LSM tree (inside one node)Copy Anchor Link

Storage engine = the code inside ONE server that puts bytes on disk and gets them back. Ours is an LSM tree = Log-Structured Merge tree: log-structured — never modify data in place, only append to logs; merge — tidy up by merging sorted files in the background. Used by Cassandra, RocksDB, LevelDB, HBase, BigTable.

Why it exists: disks are ~10× faster at sequential I/O (writing at the end of a file) than random I/O (jumping around updating bytes). B-trees — the classic engine under SQL databases — update pages in place = random I/O on every write. LSM appends only → huge write throughput.

8.1 Write pathCopy Anchor Link

1. Append to the WAL (write-ahead log) — an append-only file on disk. Pure insurance: never read in normal operation, only replayed after a crash to rebuild what RAM lost. This is the durability guarantee.
2. Insert into the memtable — an in-memory sorted map (skip list or balanced tree). Sorted so it's fast to read and trivial to dump in order.
3. Memtable hits its size threshold → frozen → written to disk in one sequential pass as an immutable SSTable; its WAL file is then deleted and a fresh memtable + WAL take over.

8.2 Inside an SSTable (Sorted String Table)Copy Anchor Link

An SSTable = an immutable (write-once, never edited) file of key-value pairs sorted by key, plus two helper blocks:

• Bloom filter: a tiny probabilistic bitmap answering "is key X in this file?" with either definitely no or maybe yes — false positives possible (~1%), false negatives never. Since a key lives in only a couple of the many SSTables, this lets reads skip almost every file without touching disk.
• Sparse index: every Nth key → byte offset of its block, so a lookup jumps straight to one block instead of scanning the file.
• Sortedness is the superpower: binary search within a file, and two sorted files merge in ONE linear pass (like merging two sorted card decks).

8.3 Reads, deletes, compactionCopy Anchor Link

• Read: memtable first, then SSTables newest → oldest (each gated by its Bloom filter). Newest-first is also why updates work: an update just writes the key again; the fresh copy is met before the stale one.
• Delete: files are immutable, so you can't erase — instead write a tombstone (a special "this key is deleted" marker). Reads that meet a tombstone answer "not found". The key + tombstone are physically removed later, at compaction. Tombstones must survive a GC grace period (longer than the repair interval) — otherwise a replica that never heard about the delete will "resurrect" the key during repair.
• Compaction: the background merge — take SSTables with overlapping key ranges, merge-sort them into new files: newest value per key wins, values shadowed by tombstones are dropped, old files deleted. Keeps the file count (and read cost) bounded and reclaims space.

8.4 The three amplifications (name the costs — the senior bit)Copy Anchor Link

Soundbite: "LSM trades read and space amplification for sequential-write throughput; B-trees make the opposite trade."

9. End-to-end paths (narrate these in the interview)Copy Anchor Link

  WRITE
  client ─► any node (becomes coordinator)
         ─► hash key onto ring → find the N replicas → stamp vector clock
         ─► send to N; each replica: WAL append → memtable insert → ack
         ─► W acks collected → tell client "success"
            (home replica down? → sloppy quorum: substitute takes it with a hint)

  READ
  client ─► coordinator ─► ask the N replicas, wait for R answers
         ─► all versions equal → return it
         ─► versions diverge → vector-clock compare → drop ancestors,
            return/merge concurrent siblings, READ-REPAIR the stale replica
         ─► inside each replica: memtable → Bloom filters → SSTables (newest first)

One-line system summary: a leaderless ring of identical LSM-backed nodes; keys placed and replicated N-ways by consistent hashing; quorums give tunable consistency; gossip + hinted handoff + Merkle repair give fault tolerance.

10. The Dynamo cheat table (memorise)Copy Anchor Link

11. Interview scriptCopy Anchor Link

  1. clarify         sizes, read/write mix, latency target, C-vs-A in a partition
  2. estimate        ~TBs, ~100k QPS → one box can't → distribute
  3. single server   hash table in RAM — then break it on capacity
  4. CAP call        "partitions happen; I'll go AP, Dynamo-style" — out loud
  5. partition       ring + vnodes (one sentence, point at CH notes)
  6. replicate       N=3 across racks; quorum R/W; the R+W>N overlap rule
  7. conflicts       LWW vs vector clocks — pick FOR the use case      ← unprompted
  8. failures        detect (gossip) / bridge (hints) / repair (Merkle) ← unprompted
  9. storage         LSM: WAL → memtable → SSTable → compaction; Bloom filters
 10. walk the paths  write end-to-end, then read; kill a node mid-walk yourself

Steps 7, 8 and the amplification costs in 9 separate a pass from a strong pass.

12. The lines that earn the mostCopy Anchor Link

1. "This is a system of record, not a cache — durability is what forces replication, quorums and a real storage engine."
2. "R + W > N guarantees the read overlaps the write on at least one node; below that I'm choosing latency over freshness — and I can dial it per request."
3. "LWW is simple but clock skew silently loses writes; vector clocks detect concurrency but push the merge to the app — I'd pick per use case."
4. "A failed node must not change the ring — gossip detects, hinted handoff bridges, Merkle-tree anti-entropy repairs."
5. "LSM trades read and space amplification for sequential-write throughput; and tombstones need a GC grace period or deleted keys resurrect."

13. Numbers to rememberCopy Anchor Link

14. Common follow-upsCopy Anchor Link

15. When NOT to go deepCopy Anchor Link

Product design interview ("design WhatsApp") → one sentence and move on:

"User/session state goes in a Dynamo-style KV store — replicated 3-way with quorum reads/writes, so it survives node failures without a leader."

Go deep only when the question IS the store: this question, distributed cache, message-broker storage, or "design DynamoDB / Cassandra" by name.

16. Glossary (A–Z, one line each)Copy Anchor Link