Design a Distributed Key-Value Store

Running example: a Distributed KV Store, a Dynamo-style store of opaque values behind put and get.

• In v1: put(key, value) and get(key)
• Values up to 400 KB
• delete writes a tombstone, it does not erase at once
• Each request picks a consistency level: ONE, QUORUM or ALL

• No joins
• No multi-key transactions
• No secondary indexes in v1

put(key, value, consistency=QUORUM)
get(key, consistency=ONE)
delete(key, consistency=QUORUM)  # tombstone

Consistency is chosen per request

• p99 read under 10 ms

• Every node runs the same software and has the same role
• No master: no leader election, no single point of failure
• Any node can act as coordinator for any key
• Coordinator is per request, not per key: the node the client hit
• Scale: 100 nodes, ~100 GB each, serving 1M reads/s + 200K writes/s

Running example: a 100-node Dynamo-style KV store holding 10 TB, ~100 GB per node.

• Job: map each key to an owner node with minimal data movement on join or leave
• Hash the key with MD5 or Murmur3 onto a 2^128 ring
• Owner = the first node clockwise from the key's position

• Store each key on N=3 nodes so data survives node and zone loss
• Preference list = coordinator + next N-1 distinct nodes clockwise
• Skip vnodes of the same host, so 3 tokens never mean 3 copies on one box
• Zone-aware: 3 replicas in 3 AZs, so one AZ outage loses at most 1 of 3 copies

• R + W > N means every read set overlaps the latest successful write set
• Standard: N=3, R=2, W=2, so any 2 replicas read must include 1 that acked the write
• Tune R and W per workload to trade read latency against write latency

• A vector clock is a list of (node, counter) pairs attached to each version
• Version A descends B if every counter in A is ≥ B's
• If neither version descends the other, they are concurrent
• Concurrent versions are kept as siblings and returned to the client to merge

Running example: Dynamo-style Distributed Key-Value Store, N=3, W=2, 200K writes/s over 100 nodes.

• Hinted handoff keeps writes available while a preference-list replica is temporarily down
• Preference list for key k: A, B, C; B is down
• Sloppy quorum: write goes to the first N healthy nodes on the ring
• So node D (next clockwise) stands in for B and the W=2 acks still arrive

Background repair of replica divergence in the Distributed KV Store (100 nodes, ~100 GB each, N=3).

• Fixes divergence that read repair and hinted handoff missed (expired hint, dead stand-in, cold keys)
• Must find differences without scanning all data
• Leaf = hash of one key range
• Parent = hash of its children's hashes
• Root = one hash that summarizes the whole token range

• Decentralized membership and liveness: no central coordinator
• Every node learns who is alive from its peers, never from a master
• Every 1 s each node exchanges state with 1-3 random peers
• Information reaches all N nodes in O(log N) rounds
• 100 nodes: about 7 rounds, so roughly 7 s to converge

Per-replica storage in the Distributed KV Store: 200K writes/s x N=3 over 100 nodes is ~6K local writes/s per node, all durable.

• Job: make local writes fast and durable on each replica
• The commit log is append-only, so the disk does sequential writes, no seeks
• Ack after the log append and the memtable insert, never after a flush

• Add a node: it takes tokens from neighbors, not from the whole ring
• Bootstrap streams only ~1/N of the data; hash mod N would move ~(N-1)/N
• Old owners keep serving until the stream ends, so no downtime