Design a distributed key-value store

Requirements

The product is deliberately narrow: store and retrieve a value by an opaque key, at scale, without downtime. Pinning that down separates what the design must do from what it is allowed to give up. Ask yourself: which of these requirements can the system bend, and which one would kill the business if it bent? The answer forces every decision that follows.

Functional

• get(key) returns the value (or values, if replicas disagree) for a key.
• put(key, value, context) writes a value; context carries version metadata the client got from a prior read.
• Values are opaque blobs up to a small cap (Dynamo used < 1 MB); the store never interprets them.
• No queries, no joins, no secondary indexes, no transactions across keys. One key at a time.

Non-functional — these are the design, not garnish:

• Always writable. A put must succeed even during failures and network partitions. This is the requirement that forces every later decision.
• Incremental scale. Add one node at a time; the cluster rebalances without a maintenance window.
• Symmetric, decentralised. No master, no special node whose loss stops the system. Every node runs the same code.
• Tunable. Different callers want different points on the latency/durability/consistency curve from the same store.

The decisive trade is availability over consistency (the A and C of CAP). Because the store must accept writes during a partition, it cannot also guarantee every reader sees the latest write — so it embraces eventual consistency and pushes conflict resolution to read time.

Estimation

Numbers decide the shape. Assume a session/cart store:

• 100M daily active users, each touching their cart ~20 times/day → 2 × 10^9 operations/day. Divide by ~10^5 s/day → ~20,000 ops/sec average, and at a 5× peak, ~100,000 ops/sec.
• The mix is read-heavy but write-real: say a 4:1 read:write ratio → ~80K reads/sec, ~20K writes/sec at peak.
• Value size ~5 KB average. Hot working set: 100M active carts × 5 KB = ~500 GB of live data — fits in the aggregate RAM of a modest cluster, so most reads can be memory-served.
• With N = 3 replication, raw stored bytes triple to ~1.5 TB before overhead; trivial on disk, the point of the number is that no single node holds it all.
• At 100K ops/sec, a single node serving ~10K ops/sec means roughly 10–20 nodes at peak with headroom — small enough to reason about, large enough that a node will fail every week.

The last figure is the real driver: at this node count, a node down is a Tuesday, not an incident. The design must treat node failure as the normal case.

High-level design

A client (or a smart client library) hashes the key onto a consistent-hashing ring, lands on a coordinator node, which replicates to the next N − 1 nodes clockwise and applies a quorum. There is no separate routing tier — any node can coordinate any request, and membership is learned by gossip.

The three building blocks, in order:

1. Partitioning — consistent hashing spreads keys across nodes and rebalances cheaply when one joins or leaves.
2. Replication + quorum — each key lives on N nodes; R and W tune how many must respond.
3. Versioning — because writes can land on different replicas during a partition, the store keeps multiple versions and reconciles them.

Deep dive

Partitioning with a hash ring

Each key is hashed onto a ring; its N replicas are the next N distinct physical nodes walking clockwise — the key’s preference list. Virtual nodes (each physical node placed at ~150 ring points) keep the arcs balanced and spread a dead node’s load across many successors instead of crushing one neighbour. The payoff is operational: adding a node moves only ~1/N of the keys, so capacity grows one box at a time with no global reshuffle. (The consistent-hashing prerequisite covers the ring mechanics; here it is the substrate the rest sits on.)

Replication and the quorum knob

Every write goes to all N replicas, but the coordinator returns success after only W of them acknowledge; every read polls N and returns after R reply. The single inequality that governs correctness is:

R + W > N   →   the read set and write set must overlap

N = 3 examples:
  W=3, R=1  → durable, fast reads, slow/fragile writes
  W=1, R=3  → fast writes, slow reads, write survives if any node is up
  W=2, R=2  → balanced; the common default

When R + W > N, any read quorum and any write quorum share at least one node, so a read is guaranteed to see at least one copy of the most recent acknowledged write. Lower the sum below N + 1 and you trade that guarantee for latency — sometimes the right call for a cache-like workload.

The catch is the word acknowledged. During a partition the coordinator may not reach the “real” N nodes, so it writes to the first N healthy nodes it can reach — a sloppy quorum — and tags the displaced writes with a hint so they get handed back when the rightful node returns (hinted handoff). This is how the store stays writable through failures: it never blocks waiting for a specific node, it just writes somewhere durable and reconciles later.

Versioning: vector clocks vs last-write-wins

Sloppy quorums and concurrent writes mean two replicas can legitimately hold different values for one key, with no global clock to order them. The store needs to know whether one version descends from the other (keep the newer) or whether they are genuinely concurrent (a conflict). Two strategies:

• Last-write-wins (LWW): attach a timestamp; the highest wins. Dead simple, but it silently drops the loser — and with clock skew across nodes, “highest timestamp” can be the wrong write. Fine for caches; dangerous for carts.
• Vector clocks: each value carries a vector of (node, counter) pairs. Comparing two vectors tells you if one happened-before the other (keep the descendant) or if they are concurrent (neither dominates) — in which case both are returned to the client to merge. Dynamo’s canonical example: two concurrent adds to a cart produce divergent vectors, and the merge is a set union, so no item is lost.

The cost of vector clocks is that the application must know how to merge — the store hands back “here are two siblings, you decide.” For a shopping cart the merge is obvious (union the items); for a counter it is not, which is exactly why purpose-built CRDTs exist. The design choice is therefore semantic: LWW where losing a write is acceptable, vector clocks where it is not.

Anti-entropy: gossip, read-repair, Merkle trees

The store has no master to track membership, so nodes run a gossip protocol: each periodically picks a random peer and exchanges its view of who is alive, which node owns which arc, and failure suspicions. Within seconds the whole cluster converges on the same map — no coordinator, no single point of failure for membership itself.

Replicas still drift (a hint never delivered, a node down during a write), so two repair mechanisms run continuously. Read-repair: when a read’s R replies disagree, the coordinator pushes the freshest version back to the stale replicas on the response path — repair piggybacks on normal traffic. Merkle-tree anti-entropy: replicas of the same key range periodically compare hash trees of their data; only the branches whose hashes differ are exchanged, so two replicas can detect and heal divergence while transferring almost nothing when they already agree.

Bottlenecks & tradeoffs

• Hot keys. Consistent hashing balances the keyspace, not load — one celebrity key (a viral cart, a global config row) routes all its traffic to the same N nodes regardless of vnodes. Mitigations: client-side caching of hot keys, splitting a hot key into shards, or bounded-load hashing for request routing. The ring does not save you from a single white-hot key.
• The merge burden. Vector clocks push reconciliation to the client; get it wrong (or default to LWW for convenience) and you silently lose data. Vectors can also grow unbounded with many coordinating nodes, so real systems prune old entries — which can, rarely, lose causal information.
• Read amplification under low R. R = 1 is fast but maximises staleness and the read-repair workload; R = N is consistent but turns one logical read into N network calls, and a single slow replica drags the tail (tail-at-scale). The R/W knob is per-operation precisely because no single setting fits every call.
• Eventual consistency is a contract, not a bug — but it leaks. “Read your own write” is not guaranteed unless you route a client’s reads through the same coordinator or use R + W > N with a strict (non-sloppy) quorum. Teams that assume read-after-write get burned; the honest design states the staleness window and lets the application opt into stronger settings where it matters.