A key-value store
that is never allowed
to say no.
Shopping carts for 300 million people, in three regions. A read that is 200 ms late costs a little money. A write that fails costs an order — so the store must accept writes while nodes are dead, while a rack is on fire, and while two halves of the cluster cannot see each other. Everything else in the design is the bill for that one promise.
The ring on the board is live. Drag a node peg around it and the key space re-divides under your hand.
Three verbs, one knob.
The surface is deliberately tiny. Every operation addresses exactly one key, and the caller chooses how much consistency it is willing to pay for on each individual call.
get(key, consistency)— returns one value, or several if the replicas disagree and the client must reconcile.put(key, value, context)— thecontextis the causal history the client last saw. It is opaque to the client and mandatory.delete(key, context)— not a removal. A delete is a write of a tombstone, because a removal cannot be replicated to a node that is currently down.- Consistency is per call: the same key can be read weakly for a product page and strongly for a checkout.
- Multi-region, with each region serving its own users locally.
Item limit 400 KB. Keys are opaque byte strings; the store never interprets them.
The budget, written down before the design.
| Objective | Target | Implied |
|---|---|---|
| p99 read, in-region | 10 ms | ≈ 2 disk seeks |
| p99 write, in-region | 15 ms | 1 WAN hop max: 0 |
| Read availability | 99.99% | 52.6 min/yr |
| Write availability | 99.995% | 26.3 min/yr |
| Durability | 11 nines | 3 AZs + 3 regions |
The stance: always writeable. When the nodes that should hold a key are unreachable, the store does not fail the write and does not block on agreement. It writes the value somewhere else, remembers where it belongs, and reconciles later. That is an explicit choice of availability over consistency under partition, and it is the single decision every later chapter pays for.
How many machines, and what actually decides it.
Two independent sizings — storage and throughput — and the answer is the larger. Move the sliders below; every figure here is recomputed.
Storage
Throughput
Replication bandwidth inside a region is a rounding error: 20 MB/s per node on a 25 Gb/s NIC — three orders of magnitude of headroom. Across regions it is not — that link is a real line item.
Why put takes something the client cannot read.
# read: may return more than one value get(key, consistency="eventual" | "quorum") -> { values: [bytes, ...], context: "opaque" } # write: context is the causal history you last saw put(key, value, context) -> { context: "opaque" } # delete: a tombstone write, not a removal delete(key, context) -> { context: "opaque" }
The context is a serialised version vector: {N1:3, N7:1},
one counter per node that has ever coordinated a write for this key. The client stores
it and hands it back untouched.
Without it the server cannot tell these two situations apart, and they need opposite handling:
| Client sends | Server can conclude |
|---|---|
| context ⊒ stored version | An update. Overwrite. |
| context ∥ stored version | A concurrent write. Keep both, return siblings on the next read. |
| no context | Nothing. Blind writes destroy other people's data. |
Opaque, because the encoding is the server's business: version vectors today, something with a smaller footprint later, without a client rewrite.
Hash the key. Give up scans.
There are exactly two ways to split a key space across machines, and they trade the same thing in opposite directions.
| Scheme | Buys | Costs |
|---|---|---|
| Range partitioning | Ordered scans, cheap prefix queries | Hot ranges — sequential keys all land on one node; needs constant splitting and rebalancing |
| Hash partitioning | Uniform spread with no operator input, forever | No scans at all. Ordering information is destroyed on purpose. |
We take hash partitioning, because the requirements already gave up scans and the operational cost of hot ranges at this scale is a full-time team. Type a key into the dock and watch it land on the ring.
key#0…key#9 and fan the reads out in the client.The hash on the board is FNV-1a/32 — deterministic, so the same key lands in the same place on every replay of this page.
hash(key) % N, then add one node.
The obvious partitioner is modulo the node count. It is uniform, it is one line of code, and it is correct — right up to the moment the cluster changes size.
The bench below holds 6 bins and 96 sampled keys. Press Add a node in the dock and watch what moves.
Nearly every key on the cluster changes owner. At our size that is nearly 19 TB of data moving to add one machine — hours of saturated network, a cold cache everywhere, and a p99 that leaves the building. The scheme is not slightly wrong; it is unusable in production for any cluster that ever grows.
Put the nodes on the ring too.
Hash the key space onto a circle. Hash each node onto the same circle. A key belongs to the first node found walking clockwise from the key's position. Nothing in that rule mentions how many nodes exist.
Add a node and it lands between two existing tokens. It takes the arc behind it from the node that used to own that arc — and only that arc. Everyone else is untouched.
But the load is lumpy. Drop N points at random on a circle and the gaps between them are exponentially distributed. The coefficient of variation of an exponential distribution is exactly 1 — so with one token per node, the spread in ownership is 100% of the mean. Some node has three times its share and someone else has almost nothing. Look at the bars on the right.
One machine, many tokens.
Give each physical node V tokens instead of one. Its share is now the sum of V independent exponential gaps, and summing independent samples shrinks the spread in a completely predictable way:
Pull the slider from 1 to 64 and watch the bars converge and the measured figure walk down the 1/√V curve. Six nodes is a small sample, so the measurement wobbles around the prediction rather than sitting on it — re-draw the tokens a few times and it will wobble around the same place. Three more things fall out for free:
- Rebuilds parallelise. A dead node's ranges are scattered across many donors, so recovery streams from V peers at once instead of from one.
- Heterogeneous hardware works. A machine with twice the disk gets twice the tokens. No special case.
- Joins are gentler. A new node bleeds a little from many nodes instead of halving one.
The preference list.
A key's coordinator is the node owning the token it lands on. Its replicas are found by continuing clockwise — but the walk has two rules that matter more than they look:
- Skip tokens belonging to a node already in the list. With V tokens per node, the next few tokens clockwise are quite likely to be the same machine. Copying a value to itself three times is not replication.
- Prefer a node in an availability zone not yet used. Three replicas in one AZ die together. With AZ awareness on, N=3 survives a full zone loss.
The cords on the board are the preference list for the key in the dock. Slack encodes role, and it survives greyscale because the shape differs, not just the ink:
| Cord | Means |
|---|---|
| Taut, thick, dark | In the quorum for this operation — actively serving |
| Slack, hanging low | In the preference list but beyond R or W — a standby copy |
| Contested, sienna, thin, tied to the wrong peg | Its replica is down; the cord is redirected to a fallback holding a hint |
R + W > N, and what it costs to buy it.
Write to N replicas, wait for W acknowledgements. Read from N, wait
for R responses. If R + W > N the two sets must
overlap in at least one node, so a read is guaranteed to see at least one copy of the
last acknowledged write. Below that, it is not guaranteed to see anything in particular.
The price is a tail latency price, and the arithmetic is order statistics, not averages. Waiting for W of N replicas means waiting for the W-th fastest, so the percentile you actually need from each individual replica is:
The jump from W=2 to W=3 at N=3 is not a 50% increase — it is the difference between needing each replica's p94 and needing its p99.7. On a real LSM node with compaction running, those are different worlds. Move the sliders and watch the modelled figures move with them.
| Setting | Character |
|---|---|
| N=3 R=2 W=2 | The default. Read-your-writes, survives one loss either way. |
| N=3 R=1 W=1 | Fastest and most available. Eventually consistent — you can read a stale cart. |
| N=3 R=1 W=3 | Fast reads, brittle writes. One node down and every write fails. |
| N=3 R=3 W=1 | Fast, durable-ish writes; reads become the fragile side. |
Sloppy quorum and hinted handoff.
A strict quorum insists the W acknowledgements come from the first N nodes on the preference list. If two of those three are down, the write fails — and we promised it would not.
A sloppy quorum takes the first N reachable nodes by walking further down the preference list. The stand-in stores the value together with a hint: metadata saying "this is not mine, it belongs to N4". When N4 comes back, the stand-in replays every hint it is holding and deletes its copy.
Kill a node in the dock. Its peg and tokens go grey, its cords turn contested and re-tie to a fallback, and a hint marker appears on the stand-in with a live count. Revive it and watch the hints walk home.
R + W > N no longer guarantees an overlap, because the W acknowledgements
may have come from nodes that are not in the read's N. Durability and availability are
preserved; the read-your-writes guarantee is not. Sloppy quorums make the store
available, not consistent — which is the trade the SLO asked for.Under a real network partition, both halves keep accepting writes for keys they can reach. Toggle the partition in the dock: the board shows what each side can still serve for R and W, and the divergence it is quietly accumulating.
Two writes, no order between them.
Client A adds a blender to the cart on one side of the partition. Client B removes a book on the other. Both writes are accepted. Neither client saw the other's version. There is no fact of the matter about which came "first".
| Last-write-wins | Version vectors | |
|---|---|---|
| Mechanism | Compare wall-clock timestamps, keep the larger | Compare per-node counters; incomparable ⇒ keep both |
| Result here | One update silently vanishes | Both survive as siblings |
| Depends on | Synchronised clocks. NTP skew of 50 ms picks a winner at random | Nothing external |
| Read cost | Always one value | Client may get several and must merge |
| Storage cost | Bounded | Siblings accumulate; needs truncation |
For a shopping cart, "merge" is a union of the add-set minus the remove-set, and the worst outcome of a merge failure is a re-added item — recoverable. For a bank balance it would not be, which is exactly why a bank balance does not belong in this store.
Modern practice: model the value as a CRDT so the merge is associative, commutative and idempotent, and the client never has to choose.
Route the write yourself.
Step 09 stated the replica-selection rule and the board has been applying it ever since. Now you write it. Your function gets the key's position on the ring, the ring itself — every virtual node the board currently holds, exactly as the sliders and pegs left it — and the replication factor N. Return the preference list: the physical nodes that will hold this write, in order. The board runs eight keys through your code and inks every verdict onto the ring.
Finding divergence without reading everything.
Hints cover a node that came back quickly. They do not cover a node that was down for a week, a dropped packet, or a disk that quietly returned the wrong bytes. Two background mechanisms close the gap:
- Read repair. A read already contacted R replicas. If their versions differ, push the merged value back to the stale ones on the way out. It is free, and it repairs exactly the data anyone is actually asking for.
- Merkle trees. Each node keeps a hash tree per token range. Two replicas compare roots — one hash. Equal means the whole range is identical and nothing is transferred. Unequal means descend, and each level halves the search.
The two trees printed on the board cover one range of 8 sub-ranges. Run the comparison in the dock: the roots differ, the walk descends the diverging branch, and exactly one leaf is transferred instead of the whole range.
Every decision, what it bought, what it cost
| Decision | Bought | Cost |
|---|---|---|
| Hash partitioning | Uniform spread with no operator input | No range scans, ever |
| Consistent hashing | Membership change moves ~1/N of keys | Ownership metadata must be gossiped and agreed |
| Virtual nodes | Spread falls as 1/√V; parallel rebuilds | V× the tokens, the gossip and the Merkle trees |
| Leaderless replication | Any node takes any write; no failover pause | No total order — conflicts become the client's problem |
| Tunable R / W | Per-call choice of latency vs freshness | Two more knobs to get wrong in production |
| Sloppy quorum + hints | Writes survive node and zone loss | R+W>N stops guaranteeing read-your-writes |
| Version vectors | Concurrency is detected, not guessed | Siblings, merge logic, bounded truncation |
| Read repair + Merkle | Divergence closes without full scans | Background I/O forever; trees to maintain |
| Multi-region async | Local latency; region loss survivable | Cross-region conflicts, and a real WAN bill |
One promise — never refuse a write — bought at the price of every row in that table. That is the whole design, and being able to name the price of each row is the difference between recognising this architecture and understanding it.