Design: a key-value store

The full lesson, with a worked example: Design: a distributed key-value store.

Requirements

Functional

• put(key, value): save a value (a cart, a session, a setting) under a key like cart:5521.
• get(key): return the value, or several conflicting versions for the app to merge.
• delete(key): remove it, so it never comes back.
• Optional: expiry times on keys; per-request choice of read and write strictness.

Non-functional

• Always writable: "add to cart" works while machines, or a whole zone, fail.
• Fast: single-digit milliseconds, even at the p99.
• Grows by adding machines: no single big machine; double the machines, roughly double the capacity.
• Tunable: each app picks its own point between fast and fresh.
• Durable: an acknowledged write survives a machine dying.
• Not required: queries by value, joins, multi-key transactions. Say so early.

Estimates

Say an app with crores of users keeps carts, sessions and settings here.

• Keys: about 100 crore (1 billion), averaging about 10 KB.
• Data: 100 crore × 10 KB ≈ 10 TB; with 3 copies, 30 TB.
• Machines for storage: about 2 TB of SSD each, kept half full for growth and housekeeping, so 1 TB a machine: about 30 machines.
• Traffic: 5 lakh requests a second at peak, 80% reads: 4 lakh reads and 1 lakh writes.
• Machine operations: reads ask 2 copies, writes go to 3: 4 lakh × 2 + 1 lakh × 3 ≈ 11 lakh a second.
• Machines for traffic: at about 20,000 operations a second each, about 55 machines.
• Plan: traffic, not storage, sets the size. Add headroom to survive losing a zone: about 80–90 machines.
• Failure rate: with that many machines, one failing every few days is routine. Design for it.

API

• PUT /v1/kv/:key with the value and the version context from the last read → 200 with the new version context.
• GET /v1/kv/:key → the value and its version context, or several sibling values when versions clash.
• DELETE /v1/kv/:key with the version context → 200; writes a tombstone.
• Per-request options: consistency=one | quorum | all for reads and writes; ttl on put.
• Client library form: get(key) returns values plus an opaque context; put(key, context, value) passes the context back so the store knows which version this write replaces.
• Rules: keys up to a few hundred bytes; values ideally under a few hundred KB, hard cap around 1 MB. Big files go to object storage with a pointer here.

Data model

• Record on a node: key, value (bytes), versionVector (a short list of node and counter pairs), writtenAt, isTombstone, expiresAt.
• Ring map: about 100–256 tokens per machine, each a point on the ring owned by one machine; shared by gossip.
• Membership table: per machine nodeId, zone, status (up, suspect, down), heartbeatCounter, lastHeardAt.
• Hint: forNode, key, value, version, stored on a stand-in until the owner returns.
• Repair summary: one Merkle tree per key range per machine, rebuilt in the background.
• On disk: a write-ahead log, an in-memory sorted table, and immutable sorted files, each with a small Bloom filter.

High-level design

• Partitioning: consistent hashing puts machines and keys on a ring; a key's home is the first machine clockwise.
• Replication: each key lives on N machines (usually 3, the replication factor): its home and the next distinct machines clockwise, in different availability zones.
• Quorums: writes wait for W confirmations, reads ask R copies.
• Coordinator: any machine can coordinate a request. It looks up the key's owners on the ring and talks to them; there is no master.
• Membership: a gossip protocol spreads who is up.
• Repair: hinted handoff, read repair and background Merkle-tree comparison.

A write with N = 3 and W = 2:

1. 1

   AppCoordinator

   PUT cart:5521 with the version it last read

2. 2

   CoordinatorNode B

   Store version 7 (B is the key's home on the ring)

3. 3

   CoordinatorNode C

   Store version 7

4. 4

   CoordinatorNode D

   Store version 7 (D is slow today)

5. 5

   Node BCoordinator

   Stored

6. 6

   Node CCoordinator

   Stored: 2 of 3, W = 2 reached

7. 7

   CoordinatorApp

   OK in about 4 ms; D catches up later

Deep dives

Partitioning: why a ring, why virtual nodes

• Mod N fails: with hash(key) mod machines, going from 10 to 11 machines keeps a key in place only about 1 time in 11, so about 90% of keys move. Terabytes cross the network to add one machine.
• Ring: positions don't depend on machine count. The new 11th machine takes over only the slice before it: about 1 in 11 keys (9%) move.
• Virtual nodes: one point per machine gives uneven slices and a newcomer relieves only one neighbour. Give each machine 100–256 points (a virtual node each), so it owns many small scattered slices, takes a little from everyone when it joins, and a bigger machine simply gets more points.
• Zone-aware placement: walk clockwise skipping machines in a zone already used, so the 3 copies land in 3 zones.
• Hot keys: one key always lands on the same 3 machines. A hot key (one product during a flash sale, 2 lakh reads a second) needs a short-lived cache in the app servers; a hot counter is split into sub-keys like likes:post9:1 to likes:post9:10, summed on read.

Replication and quorums

1. 1

   AppCoordinator

   GET cart:5521

2. 2

   CoordinatorNode C

   Read

3. 3

   CoordinatorNode D

   Read

4. 4

   Node CCoordinator

   Version 7

5. 5

   Node DCoordinator

   Version 6 (missed the write)

6. 6

   CoordinatorApp

   Version 7, the newer one

7. 7

   CoordinatorNode D

   Read repair: here is version 7

• The rule: if R + W is more than N, every read set overlaps every write set, so at least one copy read has the latest version. This is a quorum.
• N = 3, W = 2, R = 2: the usual balance. One machine down, reads and writes both still work and stay fresh.
• W = 1, R = 1: fastest, but a read can miss a recent write. Fine for view counts, wrong for "I just saved my setting".
• W = 3, R = 1: fresh, fast reads, but any one copy down blocks writes.
• Tail latency: the coordinator waits only for the second-fastest copy, so one slow machine doesn't slow every request.
• Sloppy quorums: if an owner is down, write to the next healthy machine instead (a sloppy quorum). This keeps writes flowing, but the copies may not be the true owners, so R + W more than N no longer guarantees a fresh read during failures.

Failure detection and repair

• Gossip: every second, each machine swaps its membership table with one or two random peers. News doubles each round, reaching 100 machines in about 7 rounds, a few seconds. No word from a machine for about 10 seconds marks it down.
• Hinted handoff: writes for a down machine go to a stand-in with a note; when the owner returns, the stand-in hands them back and deletes its copy. Covers short outages.
• Read repair: when a read sees an old copy, the coordinator sends it the newer version. Fixes keys that get read.
• Background repair: for everything else, machines compare copies with a Merkle tree per range. Compare the top fingerprint; if it differs, descend only into branches that differ. Ten mismatched keys among a crore are found by swapping a few thousand small fingerprints.
• Permanent loss: a dead 1 TB machine leaves its keys with 2 copies. One neighbour streaming at about 100 MB a second needs about 3 hours; with virtual nodes, about 50 neighbours share the work and finish well under an hour.

Conflicts and deletes

• Last write wins: keep the later timestamp. Simple, but clock skew of even 2 seconds can make an older write "win" and silently drop a newer one. Fine for "last seen" times.
• Vector clocks: each version carries a counter per coordinating machine. The store can tell "B came after A" from "A and B happened side by side". Side-by-side versions are kept as siblings and both returned; the app merges.
• Merge carefully: a cart that simply unions both lists brings back removed items. Track adds and removes per item.
• Tombstones: a delete writes a tombstone ("deleted at this time") that travels like any write. Keep it about 10 days; repair must reach every copy within that window, or the deleted value can come back.

Inside each machine: an LSM tree

• Write path: append to the write-ahead log for safety, insert into the memtable (sorted, in memory), acknowledge. Fast because disk writes are sequential.
• Flush: when the memtable fills (say tens of MB), write it out as an immutable sorted file, an SSTable.
• Read path: memtable first, then sorted files newest to oldest, skipping files whose Bloom filter says the key isn't there.
• Compaction: background merging of small files into bigger ones drops overwritten values and expired tombstones. It costs disk bandwidth, so leave headroom (the "half full" in the estimate).
• This is an LSM tree: very good at writes, slightly more work on reads.

Trade-offs

• Available vs consistent: under the CAP theorem, during a network split you either keep answering (and reconcile later) or refuse on one side. This design answers: right for carts, sessions, likes.
• When not to use it: balances, stock counts, anything that must never go negative. Use a leader per key range chosen by consensus, which refuses writes on the minority side of a split, or a transactional database.
• Speed vs freshness every day: even with no failures, waiting for more copies costs latency (the PACELC view). R and W let each feature choose.
• Last write wins vs versions: simple and lossy, or exact and pushes merging onto the app.
• More virtual nodes vs fewer: smoother balance and faster rebuilds, but bigger ring maps and more repair streams.
• LSM vs B-tree storage: LSM favours write-heavy loads; a B-tree favours reads and range scans.

Interview talking points

• State the API and the non-goals (no joins, no multi-key transactions) up front.
• Estimate and show that traffic, not storage, sets the machine count.
• Explain why mod N fails with the 90% versus 9% numbers, then add virtual nodes.
• Walk through N, R, W with R + W more than N, and what each setting costs.
• Cover gossip, hinted handoff, read repair and Merkle-tree repair as layers, each fixing what the previous misses.
• Contrast last write wins with vector clocks, and mention tombstones before the interviewer does.
• Make the CAP choice explicit and say which data you would not put here.

Follow-up questions

How do you add a machine without downtime? It joins the ring with its virtual nodes and streams its new slices from current owners in the background while they keep serving. Once each slice has caught up, gossip spreads the new ownership and it starts taking requests for that slice.

How is this different from a cache like Redis? A cache can lose data and rebuild it from a database. This store is the database, so it needs durable copies, repair, tombstones and careful conflict handling.

A feature needs read-your-own-writes. What do you do? Use R + W more than N for that feature (for example W = 2, R = 2), or send the follow-up read to the copy that took the write. Keep W = 1, R = 1 only where brief staleness is harmless.

What happens during a network split between zones? Both sides keep accepting writes for keys that can reach W machines, using stand-ins if needed. After the split heals, hints, read repair and Merkle repair converge the copies, and clashing versions surface as siblings for the app to merge.

Why not just run one big database with replicas? One primary caps write throughput and becomes a single point of failure during failover. The ring spreads writes across every machine and has no master to lose.

What if a feature needs strong consistency? Run consensus per key range so one leader orders every write for a key, accepting slower writes and refusals on the minority side of a split.