Notes
Requirements
Pin down the API and goals first. Put, get and delete. Keys up to a couple of hundred bytes, values up to a megabyte. Billions of keys, a hundred thousand operations a second, p ninety-nine under ten milliseconds. It must survive a node or a whole rack failing, scale by adding machines, and let callers trade consistency for latency. No range queries or transactions; that keeps the design simple.
put(key, value, [context]) get(key) → value(s), context delete(key) key ≤ 256 B, value ≤ 1 MB billions of keys, 100k+ ops/s p99 under 10 ms survive node + rack failure tunable consistency scale by adding nodes
One node
Start with one node. Most stores like this use an LSM tree, from Concepts episode twenty-three. A write is appended to a commit log for durability, inserted into a sorted in-memory table, and acknowledged. When that table fills, it's flushed as an immutable sorted file. Reads check memory, then files newest first, with a Bloom filter per file to skip the ones that can't hold the key. Background compaction merges files and drops overwritten values.
- write: append to commit log
- (durability), then memtable
- (sorted, in RAM) → ack
- memtable full → flush as an
- immutable sorted SSTable
- read: memtable, then SSTables
- newest first; Bloom filter
- skips files without the key
- compaction merges files
Partitioning
To spread keys, hash each one onto a ring, as in Concepts episode twenty-two. Each physical node owns many virtual positions, so data spreads evenly and a new node takes a small slice from everyone instead of one neighbour. To replicate, walk clockwise from the key's position and take the next three distinct physical nodes, ideally in different racks. That list is the key's preference list.
- ring of hash values (e.g.
- MD5 / Murmur, 64-bit)
- each node owns many virtual
- tokens (e.g. 16-256)
- → even spread, and a new node
- takes a little from everyone
- replicas: walk clockwise,
- take the next N distinct
- nodes (preference list)
Write path
A write goes to any node, which acts as coordinator. It forwards to all three replicas and replies once W of them confirm. Reads ask R replicas and return the newest version. With W two and R two, R plus W is greater than N, so reads overlap the latest write, as Concepts episode twenty-seven explains. Callers pick per request: quorum for a shopping cart, one for a view counter.
- N = 3 replicas
- write: send to all 3,
- reply after W acks
- read: ask R, newest wins
- W=2, R=2: R + W > N
- → reads see latest write
- W=1, R=1: fastest, eventual
- any node can coordinate
- (no leader, no failover)
Conflicts
Without a leader, two clients can write the same key on different replicas at once. Last-write-wins keeps the higher timestamp. It's simple, but it silently drops one write, and clock skew makes the choice arbitrary. Vector clocks attach a per-node counter to each value, so the store can tell 'newer' from 'concurrent' and return both versions. The client merges them, for a cart by taking the union of items.
- concurrent puts to one key on
- different replicas → conflict
- last-write-wins: timestamp
- decides; simple, may drop data
- (clock skew!)
- vector clocks: [A:2, C:1]
- detect true concurrency;
- return both, client merges
- (e.g. union of cart items)
Failure detection
How do nodes know who's alive without a central master? Gossip. Every second, each node tells a few random peers its view of everyone's heartbeat counter. News spreads to the whole cluster in a logarithmic number of rounds. If a node's counter hasn't risen for about ten seconds, peers mark it suspect. Smarter detectors like phi accrual adapt the threshold to how jittery the network normally is.
- every second each node tells
- a few random peers its view:
- node → heartbeat counter
- news reaches all in O(log n)
- rounds
- counter not rising for ~10 s
- → mark suspect / down
- phi-accrual detector adapts
- to network jitter
Hinted handoff
For short outages, use hinted handoff. Node C is down, so the coordinator writes C's copy to the next healthy node on the ring, E, tagged as belonging to C. Writes stay available. When gossip says C is back, E replays the hints and deletes them. Hints are kept for a few hours; if C stays down longer, a full repair catches it up.
- C is down: the write goes to
- the next node on the ring, E,
- tagged "belongs to C"
- writes stay available
- (sloppy quorum)
- C returns → E replays hints
- hints kept for a few hours;
- longer outage → full repair
Anti-entropy
Replicas still drift apart, from lost hints or disk problems, so a background anti-entropy process compares them. Each replica builds a Merkle tree over its key range: leaves hash buckets of keys, parents hash their children. Two replicas compare root hashes. If they match, done. If not, they descend only into the branches that differ, so they transfer a handful of keys instead of terabytes.
- each replica hashes its key
- range into a tree:
- leaves = hash of a key bucket
- parents = hash of children
- compare roots: equal → done
- differ → descend only into
- differing branches
- cost ∝ differences, not
- data size
Variants
There are two families. Leaderless, like Cassandra and Riak, which we've just designed: always writable, tunable, but conflicts are possible. Or a leader per range: the key space is split into ranges, each replicated by its own Raft group, like TiKV or CockroachDB. Every key has one leader, so reads and writes are linearizable and there are no conflicts, at the cost of a few seconds of unavailability for a range while a new leader is elected.
- leaderless (Dynamo, Cassandra,
- Riak): quorums, always
- writable, conflicts possible
- leader per range (TiKV,
- CockroachDB; DynamoDB uses
- Paxos): consensus group per
- range, linearizable,
- brief unavailability on
- leader change
Scaling
Scaling out: a new node chooses its tokens, announces itself through gossip, and streams the affected key ranges from their current owners while the cluster keeps serving. Throttle that streaming so it doesn't hurt live traffic. What adding nodes doesn't fix is a hot key, because a single key always lives on just three nodes. Put a cache in front of it or split it into several sub-keys.
- new node picks tokens, joins
- via gossip, streams its new
- ranges from current owners
- while serving continues
- throttle streaming so live
- traffic is not hurt
- hot key: one key → 3 nodes
- only; cache it in front or
- split it (key#1..#8)
Sizing
Sizing. Two billion keys at about a kilobyte each is roughly two point two terabytes. Times three replicas is six point six, and LSM compaction needs spare space, so plan about ten terabytes. With one usable terabyte per node, that's ten nodes; run twelve. Then check throughput: two hundred thousand operations a second at about twenty thousand per node is also ten. Size for both, take the larger, add headroom.
2B keys × (100 B key + 1 KB val) ≈ 2.2 TB raw × 3 replicas = 6.6 TB × 1.5 compaction headroom ≈ 10 TB on disk per node: 1 TB usable NVMe → ~10 nodes; run 12+ 200k ops/s ÷ ~20k per node → 10 nodes; same answer
Say it like this
In the interview, go layer by layer. The API and goals. One node's storage engine. Partitioning with a ring and virtual nodes. Replication to three nodes in different racks with quorum reads and writes. Conflict handling with last-write-wins or version vectors. Gossip for failure detection, hinted handoff for short outages, Merkle trees for repair. Close with disk and throughput numbers.
- API + goals (AP, low latency)
- node: commit log, memtable,
- SSTables, Bloom, compaction
- ring + vnodes, N = 3 racks
- quorum W/R per request
- conflicts: LWW or vectors
- gossip failure detection
- hinted handoff + Merkle
- repair
- numbers: disk + ops/s