A key-value store is the simplest possible database — get(k) and put(k,v) — and yet making it distributed, always-on, and durable forces every hard idea in the field into the open. We'll build a mini-Dynamo from first principles, one trade-off at a time.
There is no smaller database than a key-value store. You hand it a key, you get back a value. Two functions, that's the whole contract. And yet the moment you insist it never goes down and never loses your data, you've quietly signed up for nearly every deep idea in distributed systems.
That's why this is the canonical interview question and the canonical teaching example. Amazon's Dynamo paper, Cassandra, Riak, DynamoDB — they're all the same skeleton, and we're going to build that skeleton here. We'll start from the two-line interface, then add the machinery it takes to spread data across many machines, keep copies in sync, survive failures, resolve conflicting writes, and store bytes on disk fast.
Each section adds one capability and pays one price. By the end you'll have a system with no master, no single point of failure, and a dial you can turn between "fast" and "correct."
The contract is tiny:
get(key) — give me the value stored under this key (or "not found").put(key, value) — store this value under this key, replacing whatever was there.A key is a short unique string; a value is an opaque blob — the store doesn't care what's inside. On a single machine you'd just use a hash map and be done. The interesting word is distributed: we want one logical store spread across hundreds of machines, and we want four properties that pull against each other.
get or put should succeed even if some machines are down or unreachable. No single failure should ever return an error to the user.put says "ok," that value survives crashes, reboots, and even losing a whole machine. We get this by keeping multiple copies.The honest truth threaded through the rest of this chapter: you cannot max out all of these at once. Every section below is really about which property we spend to buy another. The first and sharpest of those trade-offs is between staying available and staying consistent.
Back in Chapter 4 we met the CAP theorem. The one-line version: when the network partitions — some machines can't reach others, which will happen — you must choose. You can keep answering requests with possibly-stale data (Availability), or you can refuse to answer rather than risk being wrong (Consistency). You cannot have both during the split, and partitions are not optional.
So it's a forced choice, and it's per-system (and, as we'll see, per-request). The two camps:
Our store leans AP — like Dynamo and Cassandra. We always want to accept writes and serve reads. But here's the elegant part we'll build toward in Section 5: we don't pick A-vs-C once and globally. We expose a tunable knob (the quorum) so each operation can slide toward "stronger consistency" or "higher availability" as needed.
No single machine can hold all the data, so we split it: each node owns a slice of the keys. The naive rule from Chapter 1's sharding was hash(key) % N. It works until N changes — add or remove one node and the modulus shifts, so almost every key suddenly maps to a different node. That's a full reshuffle of the entire dataset just to add one machine. Unacceptable for "scale smoothly."
The fix is consistent hashing (we built this in Chapter 6, so just the recap here). Imagine a circle — a ring of hash values from 0 up to the max, wrapping around. Hash each node to a point on the ring. Hash each key to a point too. A key belongs to the first node you meet walking clockwise from the key's position. That's it.
The magic is what happens when membership changes. Add a node and it slots in at one spot on the ring, taking over only the keys between it and its clockwise neighbor — roughly 1/N of the keys move, and only from one neighbor, not a global reshuffle. Remove a node and its keys flow to the next node clockwise. Nothing else is disturbed.
One refinement, used in real systems: each physical node is placed at many points on the ring (virtual nodes). This evens out the slices so no one node gets an unluckily huge arc, and it spreads a failed node's load across many survivors instead of dumping it all on one neighbor.
% N has none of it.So far each key lives on exactly one node — and if that node dies, the key is gone. That fails both "available" and "durable." The fix: store every key on more than one node. We pick a replication factor N (commonly 3) and store each key on the node that owns it plus the next N−1 nodes clockwise around the ring. Those N nodes are the key's preference list.
This composes beautifully with consistent hashing. The key's home node is found exactly as before; the replicas are just "keep walking clockwise and copy it to the next few nodes you pass." With N = 3, any single node can die and two full copies of every key it held still exist. The data outlives the machine.
For real durability you go further: place those replicas in different failure domains — different racks, different power supplies, and across different data centers. A single rack losing power, or an entire data center going dark in a flood or fiber cut, shouldn't take a key with it. Cross-datacenter replicas cost latency (the bytes have to travel) but they're what turns "durable against a dead disk" into "durable against a dead building."
Now that each key has N copies, two questions appear. When you put, how many of the N replicas must acknowledge the write before you call it "done"? And when you get, how many replicas must respond before you trust the answer? Those two numbers are the most elegant control in the whole system.
N — the number of replicas for each key (e.g. 3).W — the write quorum: how many replicas must ack a put before it succeeds.R — the read quorum: how many replicas must respond to a get before you return an answer (you take the newest version among them).Here is the one inequality to remember:
W + R > N
If W + R > N, then the set of replicas you wrote to and the set you read from must overlap by at least one node — and that overlapping node has seen your latest write. So your read is guaranteed to see it. This is "strong-ish" consistency without a master coordinating everything. (Pigeonhole principle: two sets of sizes W and R drawn from N slots, with W+R>N, can't be disjoint.)
And because W and R are just numbers, you tune them per workload:
R (fast reads), larger W. E.g. N=3, W=3, R=1: reads hit one node and are instant; writes must reach all three.W, larger R. E.g. N=3, W=1, R=3: writes ack the moment one replica has them; reads pay the cost.N=3, W=2, R=2 (sum 4 > 3). The popular default: tolerate one node down on either path and still overlap.W=1, R=1 (sum 2, not > 3). Both ops succeed if any one replica answers — but reads can be stale. This is full AP / eventual consistency.One dial, the whole CAP spectrum. Slide it per request if you like: a balance check reads with high R; a "like" button writes with W=1. Play with it below.
The quorum dial slides between named consistency models. Worth pinning down what each one actually promises, because the words get thrown around loosely.
W+R>N we get a practical approximation of this. The cost is latency and reduced availability during partitions (you might not be able to assemble the quorum).W=1, R=1 gives you. The win is enormous availability and low latency: any reachable replica answers immediately. The price is that "eventually" is doing real work — your application must tolerate temporary disagreement.What does "eventual" actually buy? Availability and latency you simply cannot get any other way. A globally-replicated, always-accepting store — a shopping cart that never refuses an item, a "like" count that's right within a second — is only possible if you let replicas diverge briefly and reconcile after. The whole AP philosophy is: accept now, agree later. Which raises the obvious question — when two replicas do disagree, how do we decide who's right?
During a partition, two clients can both put to the same key on different replicas. When the partition heals, those replicas hold two different values for one key. Which wins? Your first instinct — "the one with the later timestamp" — is a trap. Clocks on different machines drift; a write that happened first can carry a later timestamp. Worse, last-write-wins silently throws away one of the two writes, and you can't even tell whether the two writes were a genuine conflict or just one update that simply followed the other.
What we actually need is to distinguish two situations:
A vector clock detects exactly this. It's a small map from node-id to a counter: {A:2, B:1} means "node A has applied 2 updates to this key, node B has applied 1." Each time a node handles a write for a key, it bumps its own counter and attaches the resulting vector to the value. To compare two versions:
≥ the matching counter in Y. Then X has seen everything Y has, plus more — keep X, discard Y.So {A:2,B:1} vs {A:1,B:2} are concurrent (A leads on one axis, B on the other) → real conflict. But {A:2,B:1} vs {A:1,B:1} → the first descends from the second → no conflict, keep the first. Vector clocks turn "is this a conflict?" from guesswork into arithmetic.
Once you've detected a true conflict, you still have to resolve it. Two strategies: last-write-wins (simple, lossy — pick one by some tiebreaker and drop the other), or hand the siblings to the application to merge — Dynamo's shopping cart merges two carts by taking the union of items. That second strategy is the deep one: if your merge is associative, commutative, and idempotent, you have a CRDT, and conflicts resolve themselves with no coordination at all. (More on that join in the code section.)
Quorums and vector clocks handle the writes you see. But replicas can drift apart quietly — a node was down during some writes, a packet was dropped, a disk bit rotted. We need a background process that periodically asks two replicas "do we hold the same data?" and repairs any difference. That repair process is anti-entropy.
The naive way is to ship every key from one replica to the other and diff them. With millions of keys that's absurdly expensive — you'd transfer the whole dataset just to discover that maybe three keys differ. The clever way is a Merkle tree (a hash tree).
Build a tree over the key-range: each leaf is the hash of one bucket of keys; each internal node is the hash of its children's hashes; the root is a single hash summarizing everything. Now two replicas compare trees top-down:
The payoff: the amount of data you exchange is proportional to the size of the difference, not the size of the dataset. Two replicas that are fully in sync confirm it in a single comparison; two that differ in one key narrow down to that key in log-many steps. Merkle trees are how Dynamo and Cassandra make "are we still in agreement?" cheap enough to run constantly.
Every mechanism so far assumes nodes know who else is in the cluster and who's currently up. With no master, how does the cluster keep a shared picture of membership and health? The answer is gossip — the same way rumors spread. Each node periodically picks a few random peers and exchanges what it knows: "here's my view of who's alive, who I last heard from, and when." Peers merge views and pass them on. Within a few rounds, news of a node joining, leaving, or going quiet reaches everyone — no central registry, no single point of failure in the failure detector itself.
Each node also sends heartbeats; if peers stop hearing from node X for a while, gossip carries the suspicion "X looks down" around the ring until enough nodes agree to treat X as failed. Now, two flavors of failure to handle:
A node is briefly unreachable (rebooting, network blip). We don't want to block writes waiting for it. With a sloppy quorum, a write that can't reach one of its intended N replicas is instead accepted by the next healthy node on the ring, which stores it with a hint: "this really belongs to node X; hold it for them." When X comes back, the temporary holder performs hinted handoff — it ships the held writes to X and forgets them. Availability is preserved through the outage; correctness is restored after it.
If a node is truly gone (dead disk, decommissioned), its replicas no longer exist on it. The cluster repairs the ring: the surviving replicas use the Merkle-tree sync from Section 8 to rebuild the missing copies onto a replacement node (or onto the next nodes clockwise), restoring the replication factor N. The same anti-entropy machinery that catches drift also reconstructs lost replicas.
Everything above is about distribution. But each individual node still has to write bytes to a disk and read them back fast. The structure these stores reach for is the LSM-tree (Log-Structured Merge-tree), and its whole philosophy is: disks love sequential writes and hate random ones — so never write randomly.
put first appends to a write-ahead log (WAL) on disk — a pure sequential append, instantly durable. If the node crashes, the WAL replays the writes.put returns now. No disk seek, no page to update in place: fast.Over time you accumulate many overlapping SSTables. Compaction runs in the background, merging several SSTables into fewer, larger ones — dropping overwritten and deleted (tombstoned) values along the way. Because the inputs are already sorted, merging is a sequential merge-sort: more sequential I/O. Compaction is the tax LSM pays to keep read amplification under control.
Traditional databases use B-trees, which update data in place — find the right page on disk and overwrite it. Great for reads (one lookup finds the current value; no checking many files) but writes mean random disk seeks to update pages. The trade is clean: B-trees optimize reads and pay on writes; LSM-trees optimize writes (sequential appends, batched flushes) and pay on reads (amplification, compaction). A write-heavy key-value store picks the LSM side of that bargain on purpose.
Section 7 said that resolving conflicts by merging is the deep strategy. Here's why, in code. A vector clock is a Map[NodeId, Long]. Merging two of them is pointwise max — for each node-id, take the larger counter. That single operation is the key: pointwise max is the least upper bound (the "join," written ⊔) on the lattice of vector clocks. It's associative, commutative, and idempotent, which means it doesn't matter what order replicas merge in, or how many times — they all converge to the same answer with no coordination.
That's the CRDT idea in one line: if your merge is a semilattice join, conflicts resolve themselves. Below, merge is the join; descends tells you if one version dominates another; concurrent is the true-conflict test. Flip between the two languages — the shape is identical.
merge is its join. Because join is associative, commutative, and idempotent, replicas can merge in any order, repeatedly, and still converge — that's a CRDT, and it's how an AP store reconciles without ever electing a leader.We started with two functions and ended with a leaderless, fault-tolerant, tunable distributed database. Here's every piece and the one job it does:
get/put, with available + durable + scalable + fast pulling against each other.1/N of them.N nodes; spread across racks & data centers for durability.W+R>N guarantees read/write overlap; one dial across the CAP spectrum.log steps, sync only divergent subtrees.