InterviewKata
System design · Worked design D03 of 46 · Warm-up · Long form

Design a distributed key-value store

Build the storage layer itself: a store that keeps key → value pairs across hundreds of machines, keeps working when machines and whole zones fail, and lets each application choose how consistent it needs to be. This is the design behind Amazon's Dynamo paper and, in spirit, Cassandra, ScyllaDB and Riak. It ties together almost all of Level 3, and it follows the interview framework phase by phase.

1. The prompt

"Design a distributed key-value store like DynamoDB or Cassandra. Clients put and get values by key; the store must scale out by adding machines and keep working when machines fail."

There are no product features to discuss, so the interview is about mechanisms: how keys are spread over machines, how copies are kept, what "consistent" means when copies disagree, and how a cluster of hundreds of nodes heals itself every week without anyone touching it.

2. Frame the problem

What operations? Just get and put, or ranges and queries too?

Get, put and delete by key. No range scans, secondary indexes or multi-key transactions.

Pure key lookups allow hash partitioning, which spreads load evenly but rules out ordered scans.

How big are values, and how much data in total?

Values up to about 1 MB, mostly a few KB; about 500 TB of data.

Storage, not throughput, will set the node count.

When nodes or a zone fail, should writes keep working?

Yes. Being always writable matters more than every read seeing the latest write.

This is the availability-over-consistency (AP) choice that shapes everything below.

Can each request choose its consistency?

Yes: a fast, possibly stale read, or a quorum read that sees the latest acknowledged write.

We need tunable N, W and R per request.

What if two clients write the same key at the same time?

For most data the last write can win; some applications want to keep both and merge.

Versioning: timestamps for simplicity, vector clocks or mergeable types where it matters.

How fast?

p99 under 10 ms within a region, at about 2M operations a second.

One network round trip to a few replicas, and a storage engine that writes sequentially.

One region or several?

One region with three zones to start; multi-region later.

Replica placement must spread copies across zones.

3. What it must do

  1. A client can put a value under a key, choosing how many copies must confirm.
  2. A client can get the value for a key, choosing how many copies to consult.
  3. A client can delete a key.

Not now:

4. How well it must do it

TargetWhy this number
Always writable: writes succeed with any one node down, and with a whole zone down consistency or availabilityFor a shopping cart or session store, rejecting a write is worse than reconciling later.
Fast: p99 under 10 ms for reads and writes within a region speedApplications call the store several times per page.
Scale out: 500 TB and 2M operations a second, growing by adding nodes with no downtime load shapeCapacity grows constantly; big-bang migrations aren't acceptable.
Durable: an acknowledged write survives node and zone failures data safetyIt's the system of record for the applications on top.
Self-healing: routine disk and node failures need no human surviving failuresAt hundreds of nodes, something fails every week.

5. Numbers that change the design

QuantityWorking→ so we…
Data500 TB × 3 replicas1.5 PB raw; at ~4 TB usable SSD per node, ~400 nodes → so storage sets the size of the cluster
Throughput2M operations/s; one node handles ~20–50K~50–100 nodes' worth → so throughput is comfortably covered by the storage-sized cluster
Failures400 disks, a few percent failing a yearA disk dies every week or two → so repair must be automatic, normal operation
Adding a nodeHash-mod-N with N = 400 → 401Almost every key would move → so we need consistent hashing

6. The nouns

on each node   LSM tree: commit log → memtable → SSTables (+ Bloom filters), compaction
per key        key → [(value, version)]           (several versions only if writes were concurrent)
cluster state  ring: token → node, node → zone, node → status   (spread by gossip)

7. The contract

A client library (or any node, acting as coordinator) exposes three calls. The consistency level and a version context travel with them.

put(key, value, consistency = QUORUM, context?) → ok
get(key, consistency = QUORUM) → [value(s)], context      // more than one value = concurrent versions
delete(key) → ok                                          // writes a tombstone

The context carries version information (for example a vector clock), so the client can say which version its write is based on. Consistency levels: ONE (fastest), QUORUM (a majority of the N replicas), ALL.

8. First working version

Start with the simplest store that spreads data: a client library that hashes each key to one of the nodes, and nodes that store values in a local engine.

8.1 Putting a value

The client computes hash(key) mod N to pick the node and sends the write there. That node appends it to its commit log, updates its in-memory table, and acknowledges.

8.2 Getting a value

The same hash finds the same node, which reads its memtable, then its on-disk files (using Bloom filters to skip files that can't contain the key), and returns the value.

8.3 Deleting a key

The node removes the key from its engine. With one copy, that's all.

Parked for later (section 9 fixes each)

  1. Adding a node changes mod N, so almost every key moves.
  2. One copy per key: a dead disk loses data, and a down node makes keys unavailable.
  3. With copies, which one wins when two clients write the same key at once?
  4. Copies drift apart after failures; how do we find and fix differences cheaply?
  5. Who knows which nodes are alive, without a single coordinator?
  6. With copies, a plain delete can come back from the dead.

9. Make it hold up

9.1 Spreading keys so nodes can come and go scale out

ApproachWhy
✗ Badhash(key) mod NChanging N remaps nearly every key; growing the cluster means moving almost all the data.
~ GoodA consistent hashing ring: each node owns the arc up to its position; a key goes to the next node clockwiseAdding a node moves only the keys in one arc. But one position per node gives uneven arcs.
✓ GreatA ring with virtual nodes (many tokens per physical node), with bigger machines owning more tokensLoad spreads evenly, and a new node takes small slices from many others, so rebalancing is fast and gentle. Module 15 has an interactive ring.

9.2 Copies and quorums always writable · durable · fast

9.3 Concurrent writes to the same key always writable · durable

ApproachWhy
✗ BadLast write wins by each node's own clockClock skew (Module 20) picks the wrong winner, and concurrent updates silently vanish.
~ GoodLast write wins with timestamps from the client or coordinator, kept well synchronisedSimple and widely used (Cassandra does this); fine when values are overwritten whole and losing a concurrent update is acceptable.
✓ Great (when it matters)Vector clocks: return all concurrent versions ("siblings") on read and let the application merge them, or use CRDT values (counters, sets) that merge automaticallyNo update is lost silently; the original Dynamo merged shopping carts this way. It costs complexity for clients.

9.4 Finding and fixing divergence cheaply self-healing · durable

ApproachWhy
✗ BadPeriodically compare every key between replicasMoves terabytes over the network to find a handful of differences.
~ GoodRead repair only: fix stale replicas when a read notices themFree, but keys that are rarely read stay inconsistent indefinitely.
✓ GreatRead repair plus Merkle trees per token range: each replica hashes its keys into a tree; replicas compare roots, then only the branches that differFinds the differing keys by exchanging a few hashes. Try it in section 13.

9.5 Membership and failure detection without a coordinator self-healing

There's no central coordinator to fail. Each node periodically gossips its view of the cluster (who's alive, which tokens they own) with a few random peers; the information reaches every node in a few rounds. An accrual failure detector marks a node "suspected down" by how late its heartbeats are, instead of a fixed timeout. A node marked down keeps its tokens (no data moves); replicas and hints cover for it. Data moves only for planned changes: adding or removing capacity.

9.6 Deletes, and the storage engine durable · fast

9.7 The whole design after the deep dives

A write while one replica is down, then a read that repairs the stale copy, through the final design.

10. Interviewer follow-ups

A customer needs strongly consistent reads and compare-and-set.

Quorum reads and writes give read-your-writes for acknowledged writes, but not linearizable compare-and-set under concurrency. For that, run a consensus group (Raft or Paxos) per token range, as TiKV and CockroachDB do, or offer a lightweight-transaction mode that uses Paxos for those operations only, as Cassandra does. It is slower and gives up availability when a range has no majority.

One key is read 100,000 times a second.

Consistent hashing balances keys, not traffic, so that key's three replicas are overwhelmed. Let clients cache it briefly, read from any replica with consistency ONE, or split the key into several sub-keys and combine (Module 15).

Add 50 nodes to a 400-node cluster without downtime.

Each new node takes tokens from many existing nodes; data for those ranges streams to it in the background, throttled, while the old owners keep serving. Once streaming completes, ownership flips in the ring via gossip. Nothing is unavailable at any point.

Expand to a second region.

Replicate each key to N replicas in each region. Clients use LOCAL_QUORUM (a majority within their own region) for low latency, accepting that the other region may be a little behind. Cross-region conflicts use the same versioning rules.

A node is down for a week.

Hints have a time limit (say three hours), so after that they're dropped. When the node returns, anti-entropy with Merkle trees finds and streams the differences. If the node won't return, replace it: a new node takes its tokens and streams the data from the other replicas.

Why not let a single leader handle each key?

Leader-based replication gives simpler consistency, but writes stop for that range while a new leader is elected, and the leader is a hot spot. The requirement was "always writable", which leaderless quorums give. It's a trade-off worth stating.

How do clients find the right nodes?

Either any node acts as coordinator and forwards requests (one extra hop), or a "smart" client library learns the ring from the cluster and sends requests straight to the replicas (fewer hops, more complex clients). Most production drivers do the latter.

Large values, say 50 MB?

Keep the store for small values. Put large blobs in object storage and store their key in the KV store. Huge values hurt compaction, repair and tail latency for everyone.

11. When things break

FailureWhat users seeHow we detect itWhat limits the damage
One node downNothing (W and R still reachable)Gossip and the failure detectorHinted handoff; reads use the other replicas
A zone downSlightly higher latencyA third of nodes suspectedReplicas are in three zones, so two copies remain; quorums still met
Disk failureNothingNode errors; node marked downReplace the node; it streams data from its replicas
Replicas diverge (missed writes)Possibly a stale read with consistency ONERead repair notices; Merkle comparisonsRead repair and scheduled anti-entropy
Compaction falls behindSlower reads, growing disk usePending compactions, read latencyThrottle writes, add nodes, tune compaction
Clock skew with last-write-winsA newer update lostHard to see; auditsTight NTP, coordinator timestamps, or vector clocks for sensitive data

12. Wrap-up and the clock

Trade-offs we made:

The CP alternative is a consensus group per range: linearizable, but a range without a majority stops accepting writes.

What we'd watch:

At 10×:

A 45-minute interview

0–4Frame the problem (operations, AP vs CP, size)
4–9Must do, how well, numbers
9–12Nouns and contract
12–19First working version, parked list
19–40Deep dives: ring (9.1), quorums and hints (9.2), conflicts (9.3), repair (9.4)
40–45Follow-ups and wrap-up

A 60-minute interview

0–5Frame the problem
5–11Must do, how well, numbers
11–15Nouns and contract
15–23First working version
23–50Deep dives 9.1–9.6 and the final picture
50–60Follow-ups: strong consistency, hot keys, adding nodes; wrap-up
LevelWhat a strong answer shows
Mid-levelHash partitioning with consistent hashing, replication factor 3, quorum reads and writes, a basic storage choice.
SeniorW + R > N explained, hinted handoff, read repair, tombstones, LWW vs vector clocks with their costs, zone-aware replica placement.
Staff+Merkle-tree anti-entropy, gossip and accrual failure detection, rebalancing without downtime, AP vs CP per range, operations (compaction, repairs, capacity).

Common mistakes:

13. Try it, check yourself, read more

Two replicas each hold 8 keys, hashed into a Merkle tree. Click keys to make replica B's copy differ, then compare: matching hashes (green) are skipped entirely; only branches with a difference (red) are explored.

Replica A
Replica B

What is hinted handoff?

It keeps writes available during short outages and heals the missing replica afterwards.

Why are deletes written as tombstones?

Repair would otherwise copy the old value back from a replica that never saw the delete.

With Merkle trees, what do two replicas exchange to find differences?

Equal hashes mean equal subtrees, so whole branches are skipped.

Why does W + R > N matter?

The read quorum and write quorum must share a node, so the read sees the latest version.
ResourceWhy
Dynamo: Amazon's highly available key-value storeThe paper this design follows, section by section.
Cassandra's Dynamo architectureHow a production system implements the same ideas today.
Designing Data-Intensive Applications, chapters 5–6Leaderless replication and partitioning in depth.