Design a distributed key-value store

Source: https://www.hack2hire.com/question-bank/questions/system-design/69dc51e8e4441834b3f7376f
Company: Airbnb, Snowflake, Microsoft, Google, Amazon
Difficulty: Hard

Problem statement

Store and serve billions of keys with low latency, partitioned across machines and replicated to survive failures, with tunable consistency.

Requirements

We are building a Dynamo-style key-value store: billions of keys spread over hundreds of nodes, each key replicated three times, with the client choosing per request how many replicas must answer.

Functional requirements

• put(key, value), get(key), delete(key) on opaque byte keys (≤ 1 KB) and values (≤ 1 MB, typically ~1 KB).
• Per-request consistency level: ONE, QUORUM, ALL, so callers trade latency for freshness.
• Optional TTL per key; expired keys stop being returned and are eventually reclaimed.
• Concurrent writes to the same key never silently vanish: the store either resolves them by a stated rule or returns siblings for the client to merge.
• Out of scope: range scans across keys, secondary indexes, multi-key transactions.

Non-functional requirements

• Latency: p99 < 10 ms for get and put at QUORUM inside one region.
• Availability: writes keep succeeding with one node down per replica set, and through a rack loss (AP by default; CP only when the caller asks for ALL or reads and writes both at QUORUM).
• Durability: an acknowledged write survives the crash of any single node; no data loss once acknowledged by W replicas.
• Horizontal scale: add a node and it takes ~1/N of the data with no downtime and no full reshuffle.
• Self-healing: replicas converge without operators after transient failures.

Back-of-the-envelope estimation

• 10B keys × ~1 KB value ≈ 10 TB logical; × 3 replicas ≈ 30 TB; with LSM space amplification (~1.5×) ≈ 45 TB on disk.
• At 2 TB usable per node (keep disks under ~60% for compaction headroom) → ~25–30 nodes minimum; plan 60 nodes so each node also fits its read load.
• Traffic: 500K reads/s and 100K writes/s at peak. With RF=3 and quorum, each read touches 2 replicas (plus an async digest to the third), each write lands on 3 → ~1M replica reads/s and 300K replica writes/s, or ~17K reads and 5K writes per node. An SSD node handles that comfortably if most reads hit the block cache.
• Write bandwidth: 300K × 1 KB = 300 MB/s cluster-wide into WALs, ~5 MB/s per node before compaction (compaction multiplies disk writes by 10–30×).
• Metadata: 60 nodes × 256 virtual nodes = 15,360 ring tokens; the whole ring map is under 1 MB and every node holds it.

API design

Clients speak a binary RPC (gRPC or a custom protocol over persistent TCP). Smart clients cache the ring and talk straight to a replica; thin clients talk to any node, which acts as coordinator.

Put(key, value, context?, consistency=QUORUM, ttl_seconds?) -> {ok, context}
Get(key, consistency=QUORUM)  -> {values: [{value, context}], found}
Delete(key, context?, consistency=QUORUM) -> {ok}

• context is an opaque version token (an encoded vector clock). A put that echoes the context from its earlier get tells the store which versions this write supersedes. Omitting it means "blind write".
• Get normally returns one value. If replicas hold concurrent versions it returns all of them as siblings; the client merges and writes back with the merged context.
• Delete writes a tombstone (a version with no value), not a physical removal, so a stale replica can't resurrect the key.

Recovery contract: a put that times out is "unknown", not "failed" — it may have landed on some replicas and will spread via repair. Clients retry with the same context; writes are idempotent per (key, version). Errors: UNAVAILABLE (fewer than the needed replicas reachable), TIMEOUT, VALUE_TOO_LARGE.

Data model

Each node runs an LSM-tree engine (RocksDB-like), chosen because the load is write-heavy and random-key: appends are sequential, reads are served from memory or one or two SSTable seeks.

• Write-ahead log: append-only file per node. Record = {crc32, length, key, value, vector_clock, ttl, timestamp}. fsync in group commits every ~2 ms (or per write for paranoid tables).
• Memtable: in-memory skip list sorted by key, ~64 MB. When full it becomes immutable and a new one opens.
• SSTable: immutable sorted file (~64–256 MB) of 4 KB data blocks, a sparse index (first key of each block), a Bloom filter, and a footer. Files are grouped into levels L0..L6, each level 10× larger than the last.
• Bloom filter: ~10 bits per key → ~1% false positives; for 170M keys per node that is ~200 MB, kept in RAM.
• Ring metadata: token (64-bit hash) → node_id, plus node_id → {address, rack, dc, status, generation}. Stored in each node's local system table and spread by gossip; the authoritative copy of membership changes lives in a small Raft-backed config service (etcd/ZooKeeper).
• Hints table: (target_node, key, version) → value, kept on the node that stood in for a down replica.

Query stories: a read for user:42 hashes to token 0x9a…, the ring gives replicas B, C, D; on each replica the engine checks the memtable, then each SSTable whose Bloom filter says "maybe", newest first, stopping at the first hit. A write appends to the WAL, inserts into the memtable, and acks.

High-level design

flowchart LR
  C[Client] --> CO[Coordinator node]
  CO -->|"put/get to N=3"| R1[Replica A]
  CO --> R2[Replica B]
  CO --> R3[Replica C]
  R1 --> E1["WAL, memtable, SSTables"]
  CO -.->|hint if replica down| H[Hints store]
  G[Gossip] -.-> CO
  G -.-> R1
  CFG[Config service] -.->|ring changes| G
  AE[Anti entropy] -.->|Merkle compare| R2

Why the naive design breaks: one big hash map on one box loses everything on a crash, can't grow past one machine's RAM, and hash(key) mod N sharding moves almost every key when N changes. The WAL fixes the crash, consistent hashing fixes growth, and replication fixes node loss.

1. Partitioning. Keys are hashed (MurmurHash3, 64-bit) onto a ring. Each physical node owns 256 virtual nodes spread around it, so load is even and a new node takes small slices from every existing node rather than half of one neighbour's range. Replicas of a key are the next N=3 distinct physical nodes clockwise, skipping nodes in the same rack.
2. Write path. The client (or any node) finds the coordinator, the first replica for the key. The coordinator increments the vector clock with its own id, sends the write to all 3 replicas in parallel, and acks the client after W=2 succeed. The third finishes in the background.
3. Read path. The coordinator sends a full read to the fastest replica and digest reads (hash of value+version) to the others, waits for R=2. If digests match, it returns. If not, it fetches full values, picks the newest or returns siblings, and writes the winning version back to stale replicas (read repair).
4. Membership. Nodes gossip once per second with 3 random peers, exchanging heartbeats and ring versions; a ring change reaches all 60 nodes in ~log2(60) ≈ 6 rounds.
5. Failure. If a replica is down, the coordinator writes to the next healthy node on the ring with a hint ("this belongs to B"), still counting toward W (sloppy quorum). Background Merkle-tree repair catches whatever hints and read repair miss.

Deep dives

LSM storage engine: from WAL to SSTable. A write costs one sequential WAL append and one skip-list insert, so a node sustains 50K+ writes/s on one SSD. When the memtable hits 64 MB it is flushed to an L0 SSTable and its WAL segment is deleted. Reads check the memtable, then L0 files (which overlap), then one file per level L1+ (non-overlapping), skipping any file whose Bloom filter says no; with a warm block cache a point read costs 0–1 disk seeks. Compaction merges overlapping files, drops overwritten versions and tombstones older than the grace period.

Crash recovery: on restart, replay WAL segments newer than the last flushed SSTable into a fresh memtable; CRCs detect a torn final record, which is discarded (it was never acked).

Quorum replication and conflict resolution. With N=3, W=2, R=2, any read set overlaps any write set, so a quorum read sees the latest quorum-acked write — unless sloppy quorum stored it on a stand-in node, which is why this is "usually consistent", not linearizable. Concurrent writes are detected with vector clocks: each version carries {node_id: counter} entries; version A supersedes B if every counter in A ≥ B's. If neither dominates, both are kept as siblings. To bound clock growth, keep at most 10 entries and drop the oldest by timestamp.

Default to LWW with hybrid logical clocks for simplicity; offer vector-clock siblings per table for data where a lost update is a bug.

Failure detection, hinted handoff, and Merkle repair. Gossip carries heartbeat counters; a phi-accrual detector marks a peer suspect when phi > 8 (roughly, no heartbeat for several times the usual gap), so slow GC pauses don't flap membership. A suspect node is only treated as down for routing; it is not removed from the ring, because a permanent removal moves data and needs an operator or the config service. While B is down, writes for B land on E with a hint; E replays hints when gossip says B is back, and hints expire after 3 hours (after that, B is too stale and full repair takes over). For lasting divergence, each node builds a Merkle tree per token range (leaves = hashes of ~1K-key buckets, depth ~15). Two replicas compare roots, descend only into differing subtrees, and stream just those buckets. Run a full repair across the cluster at least once per tombstone grace period (e.g. weekly with 10-day grace), or deleted keys can come back from a replica that missed the tombstone.

Coordinator routing and rebalancing. Token-aware clients hash the key and send to a replica directly, saving a hop (~0.5 ms). Adding a node: the config service assigns it 256 tokens; it streams each range from current owners while also receiving new writes for those ranges (double-writes), and flips to serving reads only once streaming is done. At 60 nodes a new node pulls ~1/61 of 45 TB ≈ 750 GB; throttled to 200 MB/s that's ~1 hour.

TTL and cross-datacenter replication. TTL is stored with each value; reads treat expired values as absent, and compaction turns them into tombstones and later drops them. Across regions, use LOCAL_QUORUM (2 of 3 in the local DC) for low latency and replicate asynchronously to other DCs (RF=3 per DC); cross-DC lag is typically 100–300 ms, and conflicts resolve with the same LWW or vector-clock rule. EACH_QUORUM exists for writes that must survive a region loss, at the cost of a cross-region round trip (~70 ms+).

Trade-offs

• Leaderless quorum gives high write availability and no failover pause, but gives up linearizability; a Raft group per range (Spanner, CockroachDB, TiKV) gives linearizable reads and writes but a leader election pause of seconds and every write through one node.
• Sloppy quorum and hinted handoff keep writes flowing during failures, but a quorum read can miss a write parked on a stand-in node until the hint is replayed.
• Vector clocks never lose concurrent updates, but push merge logic onto clients and grow metadata; LWW is simple but silently drops one of two concurrent writes.
• Leveled compaction keeps reads to ~1 seek, but rewrites each byte 10–30 times, costing SSD life and disk bandwidth.
• Virtual nodes even out load and speed rebuilds, but make the ring map larger and repair ranges more numerous.
• Tombstones make deletes safe under replication, but deleted data costs space until the grace period ends, and a node offline longer than that must be rebuilt, not rejoined.
• Gossip scales without a central bottleneck but converges in seconds; a central config service is authoritative but must itself be kept highly available.

Interviewer expectations

Mid-level (L4)

• Explains why mod N hashing fails and uses consistent hashing with replication factor 3.
• Gives a clear put/get path with a WAL for durability and a coordinator fanning out to replicas.
• Knows quorum R + W > N and what ONE vs QUORUM trades.

Senior (L5)

• Walks through the LSM engine (memtable, flush, SSTables, Bloom filters, compaction) and sizes nodes and disks with numbers.
• Handles failures concretely: gossip plus phi-accrual detection, hinted handoff, read repair, Merkle-tree anti-entropy, tombstones and grace periods.
• Picks a conflict strategy (LWW vs vector clocks vs CRDTs) and explains what each loses.

Staff (L6+)

• Frames the design as a choice between leaderless quorum and per-range consensus, and says which workloads justify each.
• Covers operations: bootstrapping and decommissioning nodes with streaming and throttles, repair scheduling against tombstone grace, compaction tuning per table.
• Handles multi-region: local quorums, async replication, conflict rules, and how a region outage changes the consistency a client actually gets.