Walkthrough: a key-value store

What you are building, and what you refuse to build

"Design a key-value store" (it is the sixth chapter of Alex Xu's first volume) sounds like a data-structure question and is really an assembly question. The hash map is the easy part. The task is to make it survive the loss of machines, outgrow one machine, and stay writable while the network is broken, using parts you have already met in this course. So this chapter does not re-teach them. It names each part, links the chapter that explains it, and spends its words on the seams: the choices where one part constrains the next.

The model is the one in Amazon's Dynamo paper (SOSP 2007), which the paper pins down tightly: state is stored as binary objects identified by a key, no operation spans multiple items, there is no relational schema, and the objects are relatively small, usually under 1 MB. That scope is the whole bargain. By refusing transactions, secondary indexes and range scans you get a system where every request touches one key, so every request can be routed by a hash, and capacity grows by adding machines.

Capacity estimation

The assumptions are mine, chosen to be round; the structure of the estimate is what to copy.

• Data: 5 billion keys × 1 KB average value = 5 TB of user data. With N = 3 replicas, 15 TB stored.
• Traffic at peak: 100,000 reads a second and 50,000 writes a second.
• Fan-out into replica work. Following Cassandra's read path, a read at R = 2 asks two replicas (one returns the data, one a digest), so 100,000 × 2 = 200,000 replica reads. (The Dynamo paper's coordinator instead asks all N replicas and waits for R answers: 300,000 reads, 450,000 operations in total, about 75 nodes at the same utilization.) A write goes to all three replicas, so 50,000 × 3 = 150,000 replica writes. Total 350,000 replica operations a second.
• Node count. Take 10,000 operations a second per node as a conservative planning figure and plan to run at 60%, so 6,000 usable: 350,000 ÷ 6,000 = 58.3, so 60 nodes, 20 in each of three availability zones.
• Storage per node: 15 TB ÷ 60 = 250 GB. Throughput, not capacity, set the node count, which is typical for this workload and is why I would not pay for 4 TB disks.
• Write path per node: 150,000 ÷ 60 = 2,500 replica writes a second × 1 KB = 2.5 MB/s into the commit log. A 32 MiB commit log segment (33.6 MB) fills in about 13 seconds. At the storage chapter's assumed write amplification of 20, compaction writes about 50 MB/s per node. The compaction budget, not the 2.5 MB/s, is what you size disks for.
• Cross-zone traffic: each write sends the value to two replicas in other zones: 50,000 × 2 × 1 KB = 100 MB/s, about 0.8 Gbps.
• Growth: adding the 61st node moves 15 TB ÷ 61 = 246 GB to it. At an assumed 100 MB/s streaming throttle that takes 2,460 seconds, about 41 minutes. Scaling out is slow and should start well before you need it.

Partitioning: the ring and its settings

Hash each key onto a ring; each node owns tokens, and a key belongs to the first node clockwise from its hash. The sharding chapter works through why hash mod N fails and why virtual nodes are the real trick. The decisions that remain are quantitative. How many tokens? Cassandra 5.0's num_tokens default is 16 (older versions used 256 random tokens, and the docs say newer versions use a deterministic allocator that needs far fewer per node). For 60 nodes that is 960 tokens. What do tokens buy? When a node dies or joins, its ranges move in many small pieces to or from many neighbours instead of dumping all its load on one.

The Dynamo paper adds a decision worth knowing because it explains operational pain. Its first scheme gave each node random tokens, so the partition boundaries were set by the tokens. Moving data meant recomputing Merkle trees and archiving the key space was hard, because partitioning and placement were entangled. The paper's third scheme (Q/S tokens per node) divides the hash space into Q equal partitions and then places partitions on nodes. Placement can change without changing partitioning, a partition is one file that can be archived or shipped as a unit, and the paper reports it had the best load-balance efficiency for the same amount of per-node metadata, at a cost: changing the node count needs coordination. The takeaway for an interview is a single sentence: decouple the partition from the node that owns it, so rebalancing moves whole partitions and nothing is rehashed.

Replica placement is the next seam. The coordinator stores the key and copies it to the next N − 1 distinct clockwise nodes. With virtual nodes the next N positions can belong to fewer than N machines, so the paper builds the preference list by skipping positions until it has N distinct physical nodes, and spreads the list across data centers. In practice you extend this to racks and zones. If you remember one rule: the preference list is chosen for failure-domain diversity, not for ring adjacency.

The request path: who coordinates, and what a quorum costs

Any node can receive a request, and the key's coordinator, typically the first of the top N nodes in the preference list, runs it. The paper describes two ways to reach the coordinator. A generic load balancer sends the request to a random node, which forwards it if it is not in the key's preference list: the client needs no special library, at the cost of a possible extra hop. A partition-aware client library routes directly and skips the hop, at the cost of shipping Dynamo-specific code into every client. Either way, every node needs to know the ring, which is what gossip is for (below).

For a put the coordinator generates the new version's vector clock, writes it locally, sends it to the other replicas and succeeds when at least W − 1 others respond. For a get it asks the replicas and waits for R responses. If it gets several versions that are causally unrelated, it returns them all. The paper's own note on cost is the one to remember: the latency of a get or put is dictated by the slowest of the R or W replicas it must wait for, which is why R and W are usually set below N.

The table and the availability sum below assume strict quorums, where only the real replicas count. A sloppy quorum (Dynamo's) raises write availability, because a stand-in node can take the write, and weakens the guarantee that a read overlaps the latest write.

The availability cost is easy to compute and impressive to say. With strict quorums, assume each replica is up 99.9% of the time, independently (a generous assumption: zones, racks and bad deploys correlate failures). A write that needs 2 of 3 replicas succeeds with probability 3 × 0.999² × 0.001 + 0.999³ = 0.002994 + 0.997003 = 0.999997, about 1.6 minutes of unavailability a year. A write that needs all 3 succeeds with 0.999³ = 0.997003, which is 26.3 hours a year. Demanding one more replica costs a factor of about a thousand in unavailability, which is why (3, 2, 2) beats (3, 3, 1) for anything user-facing.

Versioning: vector clocks against last write wins

Two clients can update the same key through different coordinators during a failure. The store must either detect that or silently pick a winner, and this is the decision users feel most. The time and ordering chapter explains the clocks; the seam question is which one you attach to a value.

Riak's documentation says vector clocks were once its sole means of tracking object history, that they cannot say which value belonged to which update (so duplicate siblings appear), and that dotted version vectors, an option since Riak 2.0, fix that and limit siblings better, so it recommends them. The answer to give is therefore "causal metadata, preferably dotted, and an application merge where concurrent updates matter; timestamps where they do not."

Dry run: the paper's five versions of one key, and what a timestamp would have done

Now replay D3 and D4 with timestamps instead, where Sy's clock is 2 ms ahead of Sz's: D3 is stamped 10.002 and D4 is stamped 10.001. Last write wins keeps D3 and drops D4, and the client that wrote D4 is never told. With vector clocks the store knew the two writes had never seen each other, kept both, and let a reader merge them into D5. The cost was the merge code and the bookkeeping.

Failure handling, in three layers

Replicas will disagree, and each layer repairs on a different timescale and with different guarantees. The replication chapter has the full table; the assembly question is what you promise and when.

• Hinted handoff (seconds to hours). A node that cannot reach a replica stores the write with a hint naming the intended recipient and delivers it when that node returns. The Dynamo paper's variant is a sloppy quorum: reads and writes use the first N healthy nodes of the preference list, so a write succeeds even when two of the three real replicas are unreachable, at the price that the nodes acknowledging it may not be the ones a later read asks. Cassandra deliberately differs: a write always goes to every replica, the consistency level only says how many replies to wait for, and a hint on the coordinator is repair, not an acknowledgement (other than at consistency level ANY). Cassandra's docs also call hints best effort that do not guarantee eventual consistency, and keep them by default for a 3-hour window.
• Read repair (on the read path). In Cassandra the coordinator reads the data from one replica and digests from others; on a mismatch it fetches the full data, reconciles by timestamp and writes the winner back to the stale replicas involved, before answering. In the Dynamo paper the coordinator asks all N replicas for their versions, answers after R, and then, once the stragglers arrive, updates any node that returned a stale version, using the vector clocks. Either way it only repairs keys that are read, and Cassandra does nothing at consistency level ONE, because no digests are requested.
• Anti-entropy with Merkle trees (days). Replicas compare hash trees over the key ranges they share and stream only what differs. This is the only layer that fixes data nobody reads, and it must run often enough to beat the tombstone grace period: Cassandra's default is 10 days, so repair every node at least every 7 days.

The Merkle arithmetic is worth doing once, using the paper's traversal (two replicas compare roots, then descend only into subtrees whose hashes differ). Cassandra instead has each replica build its whole tree by scanning the range and then compares the trees, so there the saving is in what gets streamed, not in the comparison. Suppose a node's range holds 250 GB and each side builds a tree with 2¹⁶ = 65,536 leaves (an illustrative size). Each leaf covers 250 GB ÷ 65,536 ≈ 3.8 MB. Compare roots: if equal, the range is in sync after one hash. If not, descend: each level compares two child hashes, and one differing key makes exactly one path mismatch, so the comparison costs 1 + 16 × 2 = 33 hash exchanges instead of 65,536, and the repair streams one 3.8 MB leaf range rather than the whole 250 GB. The trade-off is the leaf size: more leaves mean less re-streamed data and more memory to build the tree. Dynamo notes the weakness that motivated its partitioning change: each node keeps a tree per key range it hosts, and when a node joins or leaves, many ranges change and their trees must be recalculated.

Membership and failure detection without a coordinator

There is no master to ask who is alive, so nodes tell each other. Cassandra's gossip runs every second: each node updates its own heartbeat, exchanges state with one random node, probabilistically tries nodes it believes are unreachable, and talks to a seed node if it has not already. State is versioned by a (generation, version) pair so newer information wins. Because each exchange spreads what a node knows to another, news reaches the cluster in a number of rounds that grows with the logarithm of the cluster size, so seconds for tens of nodes at one round a second (a planning estimate, not a measured figure). The Dynamo paper adds seeds: well-known nodes that every node eventually reconciles membership with, which prevents the cluster splitting into two logical rings that never learn of each other.

Deciding that a node is dead is a separate problem. Cassandra runs a variant of the phi accrual failure detector: instead of a binary timeout, each node turns the silence since the last heartbeat into a suspicion level, phi, and marks a peer down when phi reaches phi_convict_threshold (default 8, "most users should never need to adjust this"). Two seam decisions follow. First, the verdict is local: Cassandra's docs say UP and DOWN status is not propagated by gossip, so two nodes can disagree about a third, and a quorum system tolerates that. Second, the detector trades speed against false alarms. Convicting too fast during a garbage-collection pause triggers hints and re-routing for a node that was merely slow, and convicting too slowly sends requests to a dead node and adds timeouts to your p99.

The storage node: a commit log, a memtable and immutable files

Each node persists with an LSM engine, explained in the storage engines chapter; here is the path through Cassandra's version with its documented settings.

• Write path. Each write is appended to a local commit log (segments of 32 MiB by default) and inserted in the memtable, which keeps data sorted by key. A memtable flush writes a sorted, immutable SSTable with a partition index; background compaction merges SSTables, which is the write amplification the capacity estimate charged for.
• Read path. The node consults the memtable and the SSTables that may hold the key. Each SSTable has a Bloom filter over its partition keys to skip files, a partition index that maps keys to positions in the data file, and a summary that samples every 128th index entry by default (in the BIG sstable format; Cassandra 5.0's newer BTI format uses a trie-based index instead). Versions found in different files are reconciled by timestamp.
• The durability seam. The commit log defaults to periodic sync: writes are acknowledged immediately and the log is fsynced every 10,000 ms by default, so an unlucky machine crash can lose up to that window on that node. batch mode waits for the fsync, at the cost of latency. The quorum is what makes the default acceptable: an acknowledged write sits in at least two nodes' logs, and with replicas in different zones one machine failure cannot lose it. The Dynamo paper makes the same trade: it describes a write buffer that cut 99.9th-percentile latency by a factor of 5 at peak, and refines it so the coordinator picks one replica to do a durable write while still waiting only for W responses.

Dry run: a write during a failure, then three ways the dead replica catches up

The ring from the figure: nodes A to E at 100, 300, 500, 700 and 900. Key k hashes to 250, so its coordinator is B and its preference list is B, C, D. N = 3, W = 2, R = 2, Dynamo-style sloppy quorum.

Three things to say about it. The client never saw a failure: W = 2 was met by real replicas. The correctness of path B relied on the vector clock to prove v2 newer, which is exactly what timestamps cannot prove under skew. And path C is the only one that works for keys that nobody reads, which is why anti-entropy must exist and must be scheduled, even though hints and read repair usually do the work. (At t2, a Cassandra-style strict quorum keeps the hint on the coordinator B instead of E, and E is never involved.)

Follow-ups and failure modes to volunteer

• Hot keys. Partitioning spreads keys, not requests. The Dynamo paper's design assumes there are enough popular keys to spread their load; a single hot key stays on its N replicas. Answer with a cache in front, key splitting (a counter as several keys summed on read) or client-side batching.
• Deletes. A delete is a write of a tombstone; purge it only after every replica has it, which is the repair-within-grace-period rule above. Otherwise deleted data returns.
• Multi-region. Put replicas of a preference list in several data centers, as the paper describes. Cassandra additionally lets you choose per request between a local and a global quorum (LOCAL_QUORUM and EACH_QUORUM); waiting globally costs cross-region latency of 60 to 230 ms (the course's round-trip figures).
• Large values. Dynamo targets objects under 1 MB. Beyond that, store a pointer and keep the blob in object storage.
• Overload on recovery. A node that returns after hours is flooded by hint replay and repair streaming at the same moment it is cold. Throttle both.

Recognizing it in an unseen problem

• Signals: "design a key-value store", "a cache that survives restarts", "a session store", "a shopping cart that is always writable", "a distributed hash table", or any prompt with simple get and put by key at large scale and a strict availability target.
• First fork: must two readers never disagree (then a consensus-backed store, as in the consensus and replication chapters) or must the system always accept writes (then this design)?
• What to refuse: multi-key transactions, secondary indexes and range scans. Naming that these are out of scope is part of the design.
• The numbers an interviewer wants: replica fan-out (R reads and N writes per client operation), node count from throughput, write amplification, and the cost of one more replica in the write quorum.
• The pitfalls to name unprompted: last write wins and clock skew, the hint window, tombstone resurrection if repair runs too rarely, hot keys, and that R + W > N guarantees overlap only for strict quorums: a sloppy quorum breaks it, and even strict quorums are not linearizable (see the consistency models chapter).
• The follow-up you will get: "a node is down for a day, then comes back: what happens?" Walk the three layers in order, and say the hint window has expired so Merkle repair and read repair do the work.