Design Key-Value Store

What are we designing? A distributed, eventually-consistent key-value store in the style of Amazon DynamoDB and Apache Cassandra. Clients do put(key, value), get(key), and delete(key). The store runs on a cluster of commodity nodes, survives node failures and network partitions, scales horizontally by adding nodes, and lets the application tune the consistency vs availability trade-off per request.

This is the canonical 'design a NoSQL store' interview problem. The interesting parts are not the data structure (it is just a hash map) but everything around it: how do we partition data across nodes, replicate it, reconcile concurrent writes, handle node failures, and let the caller pick a consistency level.

Functional requirements.

• put(key, value) — store or overwrite a value, with optional context (vector clock) for conflict detection.
• get(key) — return the value, or a list of sibling values if there are unresolved conflicts.
• delete(key) — tombstone the key (lazy delete).
• Configurable consistency: caller picks W (write quorum) and R (read quorum) per request.

Non-functional requirements.

• Availability: writes always succeed even during a partition (AP system). If a node is reachable, it accepts the write.
• Scalability: linear horizontal scaling to thousands of nodes and petabytes.
• Latency: p99 read/write < 10 ms within a datacenter (in-memory + SSD).
• Elasticity: adding or removing a node should not require downtime and should only move a fraction of the keyspace.
• Tunable consistency: per-request, not per-cluster.

Non-goals. No multi-key transactions, no JOINs, no secondary indexes (v1). These are layered on (e.g. Cassandra's LWT and SASI) later.

Capacity estimation. Assume a 1 PB working set across a 100-node cluster.

Per node. 1 PB / 100 nodes = 10 TB/node. A modern NVMe SSD holds 8-15 TB, so one disk per node. Replication factor N=3 means each logical key is stored on 3 nodes, so raw storage is 3 PB across the cluster = 30 TB/node — still fits with 2-3 disks.

Throughput. If the cluster sustains 1M ops/sec, that's 10K ops/sec/node — easily handled by an LSM-tree storage engine (RocksDB) on SSD.

Memory. Each node caches the hot 5% of keys in RAM. 10 TB x 5% = 500 GB. That is too much for RAM; we cache only the hot 0.1% (10 GB) — fits comfortably in 64 GB RAM with room for block cache, memtable, and index.

Network. 1M ops/sec x 1 KB avg value = 1 GB/s = 8 Gbps. With RF=3, intra-cluster replication traffic is 3x writes = 24 Gbps of internal traffic if every write is synchronously replicated. Use 25 Gbps NICs and batch replication.

Key space. 64-bit hashes give 1.8 x 10^19 keys; consistent hashing maps the ring to nodes. Virtual nodes (200-500 per physical node) ensure even distribution.

APIs.

put(key, value, context=None, W=quorum, N=3)
    -> ok | conflict

get(key, R=quorum, N=3)
    -> (value, context) | [(v1, ctx1), (v2, ctx2), ...]   # siblings

delete(key, W=quorum, N=3)
    -> ok      # writes a tombstone

context is an opaque vector clock returned by the previous get. The client must round-trip it on the next put so the store can detect concurrent writes.

Consistency levels. W and R are quorum sizes:

• W=1, R=1 — fastest, weakest (any one replica answers).
• W=quorum, R=quorum — strong consistency IF W + R > N (quorum intersection).
• W=N, R=N — slowest, strongest (all replicas must agree).
• W=1, R=N — write fast, read everything (good for read-heavy + occasional conflict resolution).

The fundamental invariant: if W + R > N, reads see the latest committed write (quorum overlap). If W + R <= N, the system is eventually consistent.

Data model.

Storage per node (LSM-tree, e.g. RocksDB):

kv (
  key        BLOB,        -- the actual key bytes
  value      BLOB,        -- the value bytes
  vector_clk TEXT,        -- serialized vector clock
  timestamp  BIGINT,     -- last-write-wins tiebreaker
  tombstone  BOOLEAN,    -- lazy delete marker
  PRIMARY KEY (key, vector_clk)   -- multiple versions per key
)

Multiple siblings can coexist (different vector_clk per key); the application resolves them on read.

Cluster metadata (gossiped via a gossip protocol):

node (
  node_id       UUID,
  address       INET,
  heartbeat     BIGINT,    -- monotonic counter; if stale, node is suspected dead
  token_ranges  JSON        -- which hash ranges this node owns
)

Each node keeps a local copy of cluster state and reconciles it via gossip every 1 second with 3 random peers.

Deep dive: consistency hashing, vector clocks, and hinted handoff.

Consistent hashing. Each node maps to one or more positions on a 0..2^64 ring (hash of node_id). Each key is hashed onto the same ring; the key's replicas are the next N nodes clockwise. Adding or removing a node only moves keys between the departing node and its neighbor — most keys stay put. To avoid hotspots when nodes are few, each physical node claims 200-500 'virtual nodes' (vnodes) scattered around the ring; this guarantees +/- 10% load balance even at small cluster sizes.

Vector clocks. Each value carries a {node_id: counter} map. When a node writes a key, it increments its own counter in the map. Two writes are concurrent iff neither clock dominates the other (A > B means A has every entry of B and at least one strictly greater). If they are concurrent, the store returns BOTH values as siblings on the next get; the application reconciles them and writes a new value with a clock that dominates both. This is how Dynamo avoids throwing away writes during a partition. The trade-off: vector clocks grow unboundedly for keys that are written by many nodes; Dynamo prunes them by keeping only the last N entries per node and a timestamp.

Hinted handoff. When the coordinator tries to replicate to replica B and B is down, it picks a healthy 'hint' node H, writes the data to H with a sticky note this is for B, deliver when B returns, and replies success to the client (W quorum is satisfied by A + H). When B comes back, H delivers the buffered writes. This keeps writes available during partitions and node restarts. Hints expire after a few hours (bounded storage) — if B is gone for a day, we rely on read-repair and anti-entropy (Merkle-tree sync) to converge.

Read repair. On a read, the coordinator fetches from R replicas; if they disagree, it picks the latest, writes it back to the stale replicas, and returns the value. This self-heals skew every time a key is read.

Anti-entropy. A background process periodically compares Merkle trees of key ranges between replicas and streams any differing keys. This catches divergence that read-repair misses (keys that aren't being read).

Bottlenecks and failure modes.

• Hot key. A single key written 100K/sec lands on the same 3 replicas; they saturate. Mitigation: client-side write spreading (append a random suffix to the key, store the real key in the value); read-only replicas; or in DynamoDB, partition keys that hash to different nodes.

• Vector clock explosion. A key written by 1000 distinct nodes (e.g. a counter incremented in a partition) has a 1000-entry clock. Mitigation: prune clocks to the last 10 entries; use dotted version vectors for counters.

• Hinted-handoff disk pressure. If 5 nodes go down, the hint nodes accumulate 5 nodes' worth of writes. Mitigation: bound hint storage (e.g. 1 GB per hint node); drop oldest hints; alert and trigger repair when hints accumulate.

• Read-repair storms. After a long partition, the first read to every divergent key triggers a read-repair write, causing a write spike. Mitigation: rate-limit read-repair; do most anti-entropy in background Merkle sync.

• Gossip convergence lag. With 1000+ nodes, gossip takes 10-30 seconds to converge, during which coordinators may write to 'dead' nodes. Mitigation: the write deadline times out and the coordinator uses hinted handoff.

• Bootstrap/rebalance hotspot. Adding a new node pulls a token range from an existing node; if the range is huge, the existing node's disk and network saturate during transfer. Mitigation: stream the range in chunks with rate limiting; use vnodes so each new node drains a small slice from many existing nodes.

• Sloppy quorums. A write that goes to a hint node instead of a real replica still counts toward the W quorum. If the hint node is on a different rack, you can lose durability on a rack failure. Mitigation: rack-aware replica placement.

Scaling strategy and trade-offs.

Horizontal scale. Add nodes; vnodes automatically rebalance a slice of the ring. A 1000-node cluster with 256 vnodes/node means each new node drains 1/1000th of the keyspace from each existing node — fast, balanced, no operator intervention.

Replication factor. RF=3 is the sweet spot: tolerates 1 failure with quorum reads/writes (W=2, R=2, since 2+2 > 3). RF=5 tolerates 2 failures but triples storage and replication traffic.

Cross-DC replication. For multi-region, run RF=3 with replicas spread: 2 in the primary DC, 1 in a remote DC. This survives a DC loss with no data loss. For stronger locality, run independent RF=3 rings per DC with async cross-DC streaming — local reads are fast, but cross-DC consistency is eventual.

Storage compaction. LSM-trees accumulate SSTables; without compaction, reads slow as the number of SSTables grows. Run tiered compaction for write-heavy workloads (lower write amp), leveled compaction for read-heavy (lower read amp). Compaction is the #1 cause of latency spikes — bound its IOPS.

Trade-offs made explicit.

• We chose AP over CP — system stays writable during a partition, but two clients may write concurrently to the same key and produce siblings.
• We chose per-request W/R — caller picks speed vs consistency, but a misconfigured caller can get stale reads.
• We chose vector clocks over last-write-wins — correct under concurrent writes, but clocks grow and require pruning.
• We chose sloppy quorums + hinted handoff — writes always succeed, but a write may live on a non-replica node temporarily, complicating failure reasoning.
• We chose gossip over a coordinator service — no SPOF, but cluster state lags 10-30s, so coordinators sometimes write to dead nodes (recovered via hint).

Mental model. Think of Design Key-Value Store as a deliberate boundary in a system. The boundary exists because something becomes harder to manage when everything is done in one place: latency, scale, failure isolation, consistency, cost, or team ownership. The useful question is not “what does Design Key-Value Store mean?” but “what pressure makes this boundary worth introducing, and what new failure mode does it create?”

Numerical lens: write down traffic, payload size, read/write ratio, peak multiplier and durability target before choosing a component. The numbers should justify the architecture.

Answer this without notes: When would you choose Design Key-Value Store, and when would you intentionally avoid it? Mention at least one bottleneck it addresses, one failure mode it introduces, and one alternative. Then quantify the workload you are designing for.

Numerical sanity check

A rough capacity check: required write throughput ≈ peak writes/s × average record size. At 5,000 writes/s and 2 KB average payloads, the raw incoming data stream is about 10 MB/s before indexes, replication and overhead.