Design a Key-Value Store (Like Redis or Dynamo)

This is lesson 1 of System Design Case Studies, Part 2. The crawler in Lesson 0 needed a giant url_state table. This lesson builds the kind of store that holds it. Redis is the usual example for one node and for an in-memory cache. Dynamo (Amazon's store) and Cassandra are the usual examples for many nodes.

Words used in this lesson

• Key-value store is a database where you look up a value by its key, with no joins and no query language beyond that.
• Hash table is an in-memory structure that finds a value from a key in about one step.
• WAL (write-ahead log) is a file where every change is written first, at the end, before the change is applied in memory. After a crash you read the file again to rebuild the state.
• fsync is the call that forces the operating system to push written data to the physical disk. Without it a power cut can lose data that "was written".
• Snapshot is a copy of the whole state at one moment. Redis calls its snapshot an RDB file, and its log an AOF (append-only file).
• Partitioning (sharding) means splitting the keys over machines.
• Consistent hashing places both machines and keys on a circle (a ring), so that adding or removing a machine moves only a small share of keys. See Consistent Hashing, Visually Explained.
• Virtual node (vnode) is one machine placed on the ring at many points, to spread load evenly.
• Replication factor N is how many machines keep a copy of each key. R is how many replicas must answer a read. W is how many must confirm a write. These are a quorum setup.
• Version vector (vector clock) is a small map {node: counter} attached to a value. It tells whether one version came after another or whether they were written at the same time without seeing each other.
• Last-write-wins (LWW) means: when copies differ, keep the one with the newest timestamp and drop the others.
• Hinted handoff means: when a replica is down, another node holds the write and hands it over when the replica comes back.
• Anti-entropy is a background process that compares replicas and repairs differences. Read repair fixes differences found during a read.

1. Requirements

Functional

• put(key, value), get(key), delete(key).
• Values are small blobs (assume up to 1 MB; typical 1 KB).
• Optional: expiry (TTL), atomic counters, range scans. We skip range scans here, because hash partitioning makes them hard.

Non-functional

• Low latency: assume single-digit milliseconds at the 99th percentile.
• High availability: the store keeps accepting reads and writes when some machines fail. (This is the Dynamo choice. A store that picks consistency first, such as one using a consensus protocol, gives up writes during a split. See CAP Theorem & PACELC.)
• Scalable: add machines to add capacity, with little data movement.
• Durable: an acknowledged write survives a crash (to the degree you pay for).
• Tunable consistency: the caller picks how strict a read or write is.

2. Back-of-envelope numbers

Assumptions are marked. Method: Back-of-the-Envelope Estimation.

Data            : assume 10 billion keys, 1 KB average value + 50 B key and metadataRaw size        : 10e9 * 1.05 KB = 10.5 TBReplication N=3 : 10.5 TB * 3 = 31.5 TB on diskPer machine     : assume 2 TB usable SSD each, kept 60% full -> 1.2 TBMachines        : 31.5 TB / 1.2 TB = 26.25 -> about 27 machines, round to 30 for headroomTraffic         : assume 1,000,000 operations/s, 90% reads / 10% writesWrites          : 100,000/s, each goes to N=3 replicas -> 300,000 replica writes/sReads           : 900,000/s; with R=2 each touches 2 replicas -> 1,800,000 replica reads/sPer machine     : (300,000 + 1,800,000) / 30 = 70,000 replica operations/s

70,000 operations per second per machine is a lot for disk but fine for memory plus SSD, so I would keep hot data in memory (a cache in front of the disk files) and check with a real load test. The key insight: replication multiplies the work, so write the multiplication in the numbers.

For one node, the memory side: 10 million keys at 1 KB each is 10 GB. That fits in the RAM of one ordinary server, which is why a single Redis node can hold a surprising amount.

3. API

PUT    /v1/kv/{key}?w=2          body: raw bytes     (headers: If-Match: <version> optional)       -> 200 { "version": {"A":3,"B":1} }GET    /v1/kv/{key}?r=2       -> 200 { "values": [ { "value": "...", "version": {"A":3,"B":1} } ] }          (more than one entry means "siblings": versions that conflict; the client merges)DELETE /v1/kv/{key}?w=2       -> 204# Errors: 404 not found, 409 version conflict (if using If-Match), 503 not enough replicas for the R or W asked

Returning a version on read and sending it back on the next write is what lets the store tell "this write came after that read" from "this write is independent". This is the context idea in Dynamo (not re-read).

4. Data model

On disk, each node keeps for every key it owns:

key        -> [ (value bytes, version vector, timestamp, tombstone?) , ... ]   one entry per sibling

A tombstone is a marker that says "this key was deleted". You cannot simply remove the entry, because a replica that was down would bring the old value back later. Tombstones are removed after a long enough time (see anti-entropy).

5. High-level architecture

                       client                         |  put/get                         v                +-----------------+          any node can be the coordinator                |  coordinator    |          (it is the first node in the key's preference list)                |  node           |                +--------+--------+        hash(key) -> position on ring -> next N distinct nodes clockwise                         |         +---------------+----------------+         v               v                v     +-------+       +-------+        +-------+     | node A|       | node B|        | node C|     <- the N=3 replicas of this key     | WAL   |       | WAL   |        | WAL   |     | table |       | table |        | table |     | snap  |       | snap  |        | snap  |     +-------+       +-------+        +-------+   ring (positions 0 .. 2^32), each node appears many times (vnodes):              A   B          D           C        key K hashes to a spot; walk clockwise;          B           A        the first 3 distinct nodes (C, A, D here) hold K.              C   D

6. Deep dives

6.1 One node: hash table + WAL + snapshot

A single node keeps everything in a hash table in memory. Reads are one lookup. The problem is a crash: memory is gone. Two tools bring the data back.

1. Write-ahead log. Before changing memory, append the change to a log file. On restart, read the log from the start and re-apply every change. Appending at the end of a file is fast (sequential disk write).
2. Snapshot. The log grows forever, and replaying it gets slow. So now and then write the full state to a file (a snapshot), and then empty the log. On restart: load the snapshot, then replay the short log.

Redis offers both. Its sample redis.conf (unstable branch) says that by default it saves a snapshot after 3600 s if at least 1 change happened, after 300 s if at least 100 changes, or after 60 s if at least 10,000 changes. The append-only file is off by default (appendonly no), and when you turn it on the default is appendfsync everysec, with always and no as the other choices. This is the durability dial:

Here is the code I ran (shortened to the important parts):

class Store {  data = new Map<string, string>();  recover() {    if (existsSync(this.snap)) this.data = new Map(Object.entries(JSON.parse(readFileSync(this.snap, "utf8"))));    for (const line of readFileSync(this.wal, "utf8").split("\n")) {      if (!line) continue;      let rec; try { rec = JSON.parse(line); } catch { break; }   // torn last line after a crash: stop here      if (rec.op === "set") this.data.set(rec.k, rec.v); else this.data.delete(rec.k);    }  }  set(k, v) { appendFileSync(this.wal, JSON.stringify({ op: "set", k, v }) + "\n"); this.data.set(k, v); } // log first, then memory  checkpoint() {                       // snapshot to a temp file, rename (atomic), then empty the log    writeFileSync(this.snap + ".tmp", JSON.stringify(Object.fromEntries(this.data)));    renameSync(this.snap + ".tmp", this.snap);    writeFileSync(this.wal, "");  }}

The test: set a, b, c; checkpoint; then set b=22, delete a, set d=4; then write half a record to the log (a simulated crash during a write); then "restart" by creating a new Store on the same folder.

[ [ 'b', '22' ], [ 'c', '3' ], [ 'd', '4' ] ]log lines after snapshot: 4

The result is right: a is gone, b has its new value, d exists, and the torn record (e) was ignored. The count of 4 lines is the 3 complete records plus the torn one. Three details matter in real systems:

• Torn writes. A crash can leave half a record at the end. Real logs add a checksum (a small hash of each record) so recovery can tell a good record from a bad one. My demo only catches a record that is not valid JSON.
• The rename trick. Write the snapshot to a temp file, then rename it. A rename is atomic, so a crash leaves either the old snapshot or the new one, never a half file. (My demo skips fsync on the files, so it is not crash-safe against power loss. A real store calls fsync on the file and on its folder.)
• Snapshot without freezing. Real stores must snapshot while still taking writes. Redis uses a child process (a fork) so the child sees a frozen copy of memory while the parent continues. I did not run this part.

The cache vs database question. Redis is memory first: the whole dataset must fit in RAM. For data bigger than RAM, stores keep the data on disk in a structure made for it: an LSM tree (log-structured merge tree: writes go to an in-memory table and a log, and are later merged into sorted files), which is what Cassandra and RocksDB use. See Database Indexing: B-Trees and LSM-Trees. The WAL idea is the same.

6.2 Many nodes: partitioning with consistent hashing

The simplest rule, node = hash(key) mod N, breaks when N changes. I measured it: with 20,000 keys, going from 5 to 6 nodes under hash mod N moves 16,560 keys, which is 82.8%. Nearly everything moves.

Consistent hashing puts nodes on a ring of positions. A key is stored on the first node clockwise from the key's position. Adding a node only takes keys from its neighbour. With one point per node the spread is uneven, so each node is placed at many points (virtual nodes).

class Ring {  points: { pos: number; node: string }[] = [];  add(node: string) {                       // vnodes points per node    for (let i = 0; i < this.vnodes; i++) this.points.push({ pos: h32(node + "#" + i), node });    this.points.sort((a, b) => a.pos - b.pos);  }  prefs(key: string, n: number): string[] { // first n DISTINCT nodes clockwise    const p = h32(key);    let i = this.points.findIndex((x) => x.pos >= p); if (i < 0) i = 0;    const out: string[] = [];    for (let step = 0; out.length < n && step < this.points.length; step++) {      const node = this.points[(i + step) % this.points.length].node;      if (!out.includes(node)) out.push(node);    }    return out;  }}

I ran 20,000 keys over 5 nodes, then added a 6th node. The ideal share for the new node is 1/6 = 16.7%.

vnodes=1:   load per node min 1743 max 6479 (ideal 4000); keys moved when n6 joins: 2111 (10.6%, ideal 16.7%)vnodes=100: load per node min 3274 max 4436 (ideal 4000); keys moved when n6 joins: 3343 (16.7%, ideal 16.7%)hash(key) mod N, 5 -> 6 nodes: 16560 keys moved (82.8%)

With one point per node, the busiest node has 3.7 times the load of the quietest (6,479 vs 1,743). With 100 points it is 1.35 times (4,436 vs 3,274). The "10.6% moved" with one point is luck of where the single point landed. Do not trust it. With 100 vnodes the move was 16.7%, as theory says. Cassandra's sample config shows the same idea: num_tokens: 16 in the trunk cassandra.yaml (I read it; I remember older versions using 256, not re-checked). More vnodes spread load better, but make repair and bookkeeping heavier.

6.3 Replication and the preference list

Each key is stored on N nodes: the first N distinct physical nodes clockwise (the code skips a second vnode of a node already chosen). This is the preference list. Real systems also try to put the N copies in different racks or zones, so one rack failure does not take all copies. The first node in the list is the coordinator for that key: it takes the request and talks to the others. Any node can take a client request and forward it to the right coordinator.

Alternatives to this leaderless design: a leader per partition with followers (see Database Replication), which gives simple ordering but needs failover and an election when the leader dies.

6.4 Quorums: N, R, W

With N copies, a write waits for W acknowledgements and a read waits for R answers. The rule:

If R + W > N, every read set overlaps every write set in at least one node,so a read sees the latest acknowledged write (if there is no failure in between that breaks the overlap).

Two honest limits. First, R + W > N gives a strong chance of reading fresh data but not full linearizability (the "as if one copy" guarantee): concurrent writes, failed writes that landed on some replicas, and clock problems can still show surprises. Second, higher R or W costs latency, because you wait for the slowest of the group (the tail latency is the problem). See Consistency Models.

6.5 Conflicts: version vectors vs last-write-wins

Two clients can write the same key at the same time through different coordinators. Both writes succeed (we chose availability). Later, replicas hold two different values. Which is right?

Last-write-wins. Attach a timestamp to each write, and keep the highest. It is simple and keeps one value. But machine clocks differ by milliseconds or more, so "last" may be the wrong one, and one of the two writes is silently lost. Cassandra uses timestamps this way (from general knowledge).

Version vectors. Each value carries {node: counter}. The coordinator increases its own counter on every write. Compare two vectors a and b:

• a dominates b if every counter in a is at least that in b: a came after b, so drop b.
• If neither dominates, the writes were concurrent: keep both as siblings and let the client merge them on its next write. The shopping cart example in the Dynamo paper merges carts so an item is not lost (not re-read).

I ran this scenario. The ring has nodes A to E, N=3, R=2, W=2. The preference list for cart:42 came out as [C, A, D].

-- 1. normal write then read (N=3, R=2, W=2)write ok: trueread: [{"value":"milk","vv":{"C":1}}]-- 3. two clients write at the same time without seeing each other (concurrent)siblings returned: [ 'milk+eggs+bread {"C":3}', 'milk+eggs+juice {"C":2,"A":1}' ]after client merges and writes back: [{"value":"milk+eggs+bread+juice","vv":{"C":4,"A":1}}]

Both concurrent carts came back as siblings, because {C:3} and {C:2,A:1} each beat the other on one counter. The client merged them (union of items) and wrote back with the merged vector {C:3,A:1}, which dominates both. The new version {C:4,A:1} replaced them. Nothing was lost. With LWW one of the two items would have vanished.

The cost: the client must write merge code, and vectors can grow if many different nodes write the same key. Dynamo cuts the oldest entries when a vector gets long (not re-read). A middle path is a CRDT (a data type built so that merging always works, for example a set that only grows). See Lesson 3 for the idea.

6.6 Failures: hinted handoff, read repair, anti-entropy

Hinted handoff. If a node in the preference list is down, a write could fail. Instead the coordinator sends that copy to the next healthy node on the ring (a stand-in), with a hint: "this belongs to D". When D returns, the stand-in sends the data back and deletes its copy. This is a sloppy quorum: the W acknowledgements may come from stand-ins, so the write succeeds, but a read of the true replicas may not see it until the handoff is done. Cassandra has hinted_handoff_enabled: true and max_hint_window: 3h by default in the trunk config: it stops storing hints for a node that has been dead longer than 3 hours (then the node must be repaired in other ways).

My run, with D down:

-- 2. one replica down: still works; the hint is stored on a stand-in nodewrite ok: truehints: [ 'E holds a copy for D' ]after D returns and hints replay, it has: [{"value":"milk+eggs","vv":{"C":2}}]

Read repair. When a read asks R replicas and sees that one is behind, the coordinator sends the newest version to the stale one. In the code it writes the winning versions back to all replicas that took part. Hot keys heal quickly. Cold keys may never be read, so we need the next tool.

Anti-entropy. A background job compares the data of replicas and copies what is missing. Comparing everything is too slow, so Dynamo uses Merkle trees (not re-read): each replica builds a tree of hashes over key ranges, where a parent hash covers its children. Two replicas compare the top hash; if equal, the ranges match; if not, they walk down and compare only the ranges that differ. Only the differing keys are transferred. The same repair also deletes tombstones safely, after all replicas have seen them.

Too many failures. If fewer than R replicas are alive, a read must fail rather than guess. My run, with two of three replicas down and R=2:

-- 4. two of three replicas down: a read needs R=2 live replicas, so it fails (better an error than a wrong answer)read with 1 live replica (R=2): null

6.7 Cluster membership

How does each node know who is in the ring? Dynamo uses a gossip protocol (each node regularly tells a few random nodes what it knows, so the news spreads in a few rounds; not re-read). A node that does not answer is marked as suspected-down, and the system treats short failures (a restart) differently from permanent ones (a removed machine), so it does not move data around for every blip. Adding a node: it takes its vnode positions, and the neighbours stream the matching keys to it.

7. Failure modes and trade-offs

Trade-offs to say aloud

• Availability vs consistency: this design chooses to answer, and fixes conflicts later. A bank balance wants the other choice.
• R and W: pick them per use case, not one number for the whole store.
• LWW vs vectors: LWW is simple and can lose data. Vectors keep data and push work to the client.
• Memory vs disk: all-in-memory is fastest and costs most per GB.
• vnodes: more points give even load but more repair bookkeeping.

8. How to say it in an interview (60 seconds)

"First the single-node core: a hash table in memory, a write-ahead log so every change survives a crash, and periodic snapshots so the log stays short. For many nodes I partition with consistent hashing and virtual nodes, so adding a node moves only about 1/N of the keys. I measured 16.7% moved when going from 5 to 6 nodes, against 83% with hash-mod-N. Each key lives on N=3 nodes, the first three distinct nodes clockwise. I use quorums: R plus W greater than N, usually 3, 2, 2. If two clients write at once, I keep version vectors and return siblings, or use last-write-wins if losing a write is fine. When a replica is down, a stand-in node keeps a hint and hands it back later. Read repair and Merkle-tree anti-entropy fix drift in the background. If fewer than R replicas are alive, the read fails. For the numbers: 10 billion keys of 1 KB is 10.5 TB, 31.5 TB with three copies, about 30 machines."

Run it yourself

kv-single.ts is run as node kv-single.ts ./somefolder, and ring.ts as node ring.ts. Both use only Node's standard library. If you copy the code, note that Node's built-in type stripping does not accept TypeScript constructor parameter properties (constructor(public x: number)), so I assigned fields by hand.

Interview questions

Key takeaways

• One node: hash table + write-ahead log + snapshots. The fsync policy is the durability dial.
• Many nodes: consistent hashing with virtual nodes. Measured: 16.7% of keys move when a 6th node joins, against 82.8% with hash mod N.
• N, R, W: R + W > N makes reads overlap with writes. 3-2-2 is the usual balance. It is not full linearizability.
• Conflicts: version vectors keep both sides as siblings. Last-write-wins is simple and loses data.
• Repair: hinted handoff, read repair, anti-entropy. A read must fail when fewer than R replicas are alive.
• Count the multiplication by N in your numbers.

What's next

Lesson 2: Design a File Sync Service Like Dropbox. The same ideas (hashes, replicas, conflicts) now work on files: cut files into chunks, store each chunk once by its hash, and sync only what changed.