YeetCode
System Design · Core Designs

Design a Key-Value Store: Consistent Hashing, Quorums, and Conflict Resolution

How to design a distributed key-value store like DynamoDB — consistent hashing, N/R/W quorums, vector clocks, Merkle-tree recovery, and LSM storage engines.

11 min readBy Bhavesh Singh
key-value storeconsistent hashingquorum consistencyvector clocksdynamodblsm-tree

This article has an interactive companion

Don't just read it — step through it interactively, one state change at a time.

Practice this topic on YeetCode

A key-value store looks trivial from the outside: three methods — get(key), put(key, value), delete(key) — and no joins, no schema, no query planner. The entire difficulty lives underneath, in what happens when you spread those three methods across 300 commodity servers in three data centers and a network link dies mid-write.

This is the Amazon Dynamo problem, and it is a favorite in interviews precisely because the simple API forces every hard distributed-systems decision into the open. There is nowhere to hide. You will partition data, replicate it, keep replicas roughly in sync, resolve conflicting writes, and let a crashed node rejoin — all while promising single-digit-millisecond latency.

The design below builds a Dynamo-style store from those requirements outward, then traces one concrete write and read through the ring so the moving parts stop being abstract.

What the store actually has to promise

Start by writing down promises, not boxes. A functional requirement says what a caller can ask for; a non-functional requirement names a bound the system must hold.

The functional surface is tiny: get(k) returns the value (or values) for a key, put(k, v) stores it, delete(k) removes it. Values are opaque blobs, typically capped at something like 1 MB. No range scans, no secondary indexes — the moment you add those, you are designing a different system.

The non-functional bounds are where the real work is:

RequirementTargetWhy it drives the design
Availability"Always writeable" — no single point of failurePushes you toward AP and leaderless replication
Latencyp99 read/write under ~10 msRules out cross-DC synchronous quorums on the hot path
ScalabilityLinear with commodity nodesForces consistent hashing over static sharding
Fault toleranceSurvive node and rack lossReplication factor N ≥ 3
ConsistencyTunable per operationN/R/W knobs, not a fixed guarantee

That last row is the design's spine. You are not building "a consistent store" or "an available store" — you are building a store where the caller dials the trade-off per request.

Capacity: why the numbers change the shape

Estimates do not need false precision; they need assumptions you can defend. Take a user-data workload — shopping carts, session state, profiles.

  • 100 million daily active users, ~20 key operations each = 2 billion ops/day.
  • 2e9 / 86,400 ≈ 23,000 ops/sec average, and a 3× peak factor ⇒ ~70,000 ops/sec at the top of the day.
  • 500 billion stored keys × ~1 KB average value = ~500 TB of primary data.
  • At replication factor N = 3, that is ~1.5 PB on disk before compaction overhead.

Two conclusions fall straight out of these numbers. First, 70k ops/sec across 1.5 PB cannot live on one machine or even ten — you need hundreds of nodes, which means partitioning is not optional. Second, at that write rate the per-node storage engine has to favor writes, which will decide the B-tree-vs-LSM question later.

Partitioning: consistent hashing, not modulo

The naive way to spread 500 TB across N nodes is node = hash(key) % N. It works until N changes. Add one node and the divisor shifts from N to N+1, so almost every key now maps somewhere new — a near-total reshuffle of 1.5 PB every time you scale or lose a machine. That is unusable.

Consistent hashing fixes the blast radius. Hash the node IDs and the keys onto the same circular space (say 0 to 2^128). A key is owned by the first node found walking clockwise from the key's position. Now adding or removing a node only remaps the keys in one arc — roughly 1/N of the data — instead of all of it. The neighbors on either side absorb the change; everyone else is untouched.

Plain consistent hashing has one weakness: with few nodes the arcs are uneven, so load is lumpy, and a heterogeneous fleet (some big boxes, some small) can't be weighted. The fix is virtual nodes: each physical machine claims many tokens scattered around the ring — often 100–200 of them. More tokens smooth the distribution and let a beefier server simply hold more tokens.

Replication and the N/R/W dial

Owning a key isn't enough; you replicate it to the next N-1 distinct physical nodes clockwise on the ring, skipping virtual-node duplicates so all N copies land on different machines (and ideally different racks). N is the durability knob.

Reads and writes then use quorums:

  • N = number of replicas (e.g. 3).
  • W = replicas that must acknowledge before a write is confirmed.
  • R = replicas queried before a read returns.

The rule that matters: when R + W > N, every read overlaps the latest write on at least one replica, so a read can always find the newest version. That gives strong consistency without a leader.

Config (N=3)R + WCharacter
R=2, W=24 > 3Balanced — strong-ish, the common default
R=1, W=12 < 3Max availability + lowest latency, eventual consistency
R=3, W=14 > 3Fast writes, slow/consistent reads
R=1, W=34 > 3Fast reads, slow durable writes

Because writes only wait for W acks (not all N), a slow or dead replica does not block the caller. The coordinator still sends to all N; it just returns as soon as W come back and reconciles the laggards later.

Walkthrough: one write and one read on the ring

Concrete beats abstract. Ring with nodes A, B, C, D, E; configuration N = 3, W = 2, R = 2. A user updates their cart under key cart:8021.

The write — put("cart:8021", {items:[socks]}):

StepWhat happensState
1hash("cart:8021") lands between D and A on the ringCoordinator = A (first node clockwise)
2A picks the N=3 preference list: A, B, C3 replicas targeted
3A writes locally, tags version with vector clock [A:1], forwards to B and CA done
4B acks. That is W=2 (A + B)Write confirmed to client
5C's ack arrives 40 ms later, out of bandAll 3 now hold [A:1]

The client got its "OK" after two acks. C caught up asynchronously — the tail replica never sat on the critical path.

The read — get("cart:8021") after a network blip left C stale:

Suppose during a partition a second write hit the same key through coordinator B, producing version [A:1, B:1] on A and B, while C still holds the old [A:1].

StepWhat happensResult
1Coordinator queries R=2 replicas: A and CTwo versions come back
2Compare vector clocks: [A:1,B:1] vs [A:1][A:1,B:1] dominates — it is strictly newer
3Return [A:1,B:1]; push it back to CRead repair heals the stale replica

Vector-clock comparison did the reconciliation. C didn't need a full re-sync — the read itself repaired it.

Conflict resolution: when neither version wins

The clean case above had one clock dominating the other. The hard case is two concurrent writes — say the cart edited on a phone and a laptop during a partition — producing [A:1, B:1] and [A:1, C:1]. Neither vector clock dominates; they are causally concurrent.

The system cannot silently pick one without losing a user's change. Two strategies:

  • Last-write-wins (LWW): keep the version with the higher wall-clock timestamp. Simple, but clock skew across machines means it can silently drop the earlier-but-actually-later edit. Fine for "last seen" flags, dangerous for carts.
  • Return both to the application: hand both siblings to the client and let domain logic merge them — for a cart, that's a union of items, which is why "add socks on phone, add shoes on laptop" ends up with both. This is Dynamo's default and why its API can return multiple values.

Failure recovery: getting a node back cheaply

A node dies for 30 minutes, then returns. Naively re-shipping its entire partition — tens of terabytes — would saturate the network for hours. Two mechanisms avoid that.

Hinted handoff keeps writes flowing during the outage. When a target replica is unreachable, a neighbor accepts the write with a "hint" that it belongs to the down node, stores it aside, and forwards it once the node returns. Availability never dips.

Merkle trees make catch-up proportional to the difference, not the data. Each replica keeps a hash tree over its key range: leaves hash key-value pairs, parents hash their children. Two replicas compare root hashes first — if they match, the ranges are identical and nothing transfers. If they differ, they descend only into mismatched subtrees until the divergent keys are pinned down. Anti-entropy becomes O(differences) instead of O(total data), which is the difference between syncing a few thousand keys and re-shipping a partition.

Storage engine: LSM-tree over B-tree

Back to that 70k ops/sec write rate. Per node, the engine choice is B-tree vs LSM-tree.

A B-tree updates records in place, which means random disk I/O on every write and better read latency. An LSM-tree (RocksDB, LevelDB) appends writes to an in-memory memtable, flushes it to immutable, sorted SSTables, and merges those in the background via compaction. Writes become sequential I/O — far higher throughput on the same disk.

For a write-heavy Dynamo-style store, LSM wins. Its cost is read amplification: a key might live in any of several SSTable levels, so a read may probe multiple files. Bloom filters cut most of that — a per-SSTable probabilistic membership check skips files that certainly don't hold the key — and a block cache absorbs the hot set. You trade a little read complexity for a lot of write throughput, which matches the workload.

How this shows up in interviews

Interviewers rarely ask you to "design a key-value store" and leave it there. They drive to the boundaries. Expect the sequence to go: sketch the ring, then "what happens when I add a node?" (consistent hashing, 1/N moves), then "how do you not lose a write when a replica is down?" (W < N plus hinted handoff), then "two people edit the same cart during a partition — now what?" (vector clocks and sibling merge), then "a node was gone for an hour, how does it catch up?" (Merkle trees).

The hot-partition follow-up is almost guaranteed: one celebrity key or one bad partition key overwhelms a single shard while the fleet sits idle. Have answers ready — adaptive splitting to isolate the hot key onto dedicated capacity, a read-through cache in front, request coalescing, or client-side write sharding by appending a random suffix and fanning reads across the shards. Name the CAP position (AP, eventual consistency, reconcile later) out loud early; it frames every later trade-off.

Further learning

FAQ

Why is consistent hashing better than hash(key) % N for sharding?

hash(key) % N ties every key's location to the node count. Change N — add a node, lose one — and the divisor changes, so nearly every key remaps and you shuffle almost the entire dataset. Consistent hashing places nodes and keys on a shared ring and assigns each key to the next node clockwise, so growing or shrinking the cluster only reassigns the keys in one neighboring arc, about 1/N of the data. Virtual nodes (many tokens per machine) then smooth out the load and let unequal hardware carry proportional shares.

What does R + W > N actually guarantee?

With N replicas, a read touching R of them and a write touching W of them are guaranteed to share at least one replica whenever R + W > N — that overlapping replica holds the most recent write, so the read can always find the newest version. It gives you strong consistency without electing a leader. N=3, R=2, W=2 is the balanced default. Drop to R=1, W=1 and R+W=2 ≤ 3, so a read can miss the latest write entirely: that's the eventual-consistency, maximum-availability setting.

How does a Dynamo-style store resolve two conflicting writes?

Each version carries a vector clock — a map of node ID to counter — incremented by whichever node coordinates a write. On read, the system compares clocks. If one clock dominates the other (every counter greater or equal), it is strictly newer and wins automatically. If neither dominates, the writes were concurrent, and the store returns both siblings to the application to merge with domain logic, such as unioning the items in a shopping cart. Last-write-wins by timestamp is simpler but can silently discard a real update when server clocks drift.

How does a recovered node catch up without copying everything?

Two mechanisms. Hinted handoff keeps availability up during the outage: a neighbor temporarily accepts writes meant for the down node and forwards them when it returns. For the data that diverged while it was gone, replicas use Merkle trees — hash trees over their key ranges. They compare root hashes, and only descend into subtrees whose hashes differ, isolating the exact divergent keys. Sync cost becomes proportional to the differences rather than the full partition, turning an hours-long re-ship into a targeted exchange of a few thousand keys.

Why choose an LSM-tree storage engine over a B-tree here?

Dynamo-style stores are write-heavy, and LSM-trees turn writes into cheap sequential I/O: append to an in-memory memtable, flush to immutable sorted SSTables, and merge them in the background via compaction. That yields far higher write throughput than a B-tree, which updates in place and pays random I/O on every write. The LSM trade-off is read amplification — a key may sit in several SSTable levels — but bloom filters skip files that can't contain the key and a block cache holds the hot set, so reads stay fast enough while writes scale.

Make it stick: run this one yourself

Don't just read it — step through it interactively, one state change at a time.

Practice this topic on YeetCode