Design a Key-Value Store

A key-value store (non-relational DB) stores unique identifier keys with associated values. Keys: plain text (last_logged_in_at) or hashed (253DDEC4). Values: strings, lists, objects (treated as opaque). Examples: Dynamo, Memcached, Redis.

Operations: put(key, value), get(key)

Step 1 of 4: Understand the problem and establish design scope #

Goal: lock in requirements before designing.

Requirements:

1. Small key-value pairs: < 10 KB
2. Ability to store big data
3. High availability (responds quickly during failures)
4. High scalability (supports large datasets)
5. Automatic scaling (add/remove servers based on traffic)
6. Tunable consistency
7. Low latency

Single server key-value store #

Store in a hash table in memory. Optimizations: data compression, keep hot data in memory + rest on disk. A single server reaches capacity quickly → need a distributed key-value store.

Step 2 of 4: Propose high-level design and get buy-in #

Goal: choose your CAP tradeoff, partitioning, and replication strategy.

CAP theorem #

A distributed system can simultaneously provide only two of three guarantees:

1. Consistency (C): all clients see the same data at the same time regardless of node
2. Availability (A): any client gets a response even if some nodes are down
3. Partition Tolerance (P): system continues operating despite network partitions

System classifications:

• CP: Consistency + Partition Tolerance, sacrifice availability (bank systems: error if inconsistency)
• AP: Availability + Partition Tolerance, sacrifice consistency (may return stale data)
• CA: cannot exist in the real world: network failure is unavoidable

Real-world: partitions cannot be avoided. When a partition occurs (n3 down), you must choose CP (block writes to n1/n2) or AP (accept reads/writes, sync later).

Data partition #

Use consistent hashing (Chapter 5):

1. Servers placed on hash ring (s0–s7)
2. Key hashed onto the same ring, stored on the first server clockwise
3. Automatic scaling: servers added/removed based on load
4. Heterogeneity: virtual nodes proportional to server capacity

Data replication #

Data replicated asynchronously over N servers. After mapping a key to its ring position, walk clockwise and choose the first N unique servers.

N = 3 example: key0 replicated at s1, s2, s3.

With virtual nodes, ensure physical uniqueness (first N on ring may map to < N physical servers). Place replicas in distinct data centers for reliability.

Step 3 of 4: Design deep dive #

Goal: consistency, conflict resolution, and failure handling.

Consistency: quorum consensus #

1. N = number of replicas
2. W = write quorum size (writes acknowledged by W replicas = successful)
3. R = read quorum size (reads wait for at least R replicas)

W = 1 doesn't mean data written to 1 server: it means the coordinator needs only 1 ack. The coordinator acts as proxy between client and nodes.

Configurations:

1. R = 1, W = N → optimized for fast read
2. W = 1, R = N → optimized for fast write
3. W + R > N → strong consistency guaranteed (at least one overlapping node has latest data)
4. W + R ≤ N → strong consistency NOT guaranteed

Typical strong consistency: N = 3, W = R = 2.

Consistency models:

1. Strong consistency: read returns the latest write: achieved by blocking reads/writes until all replicas agree. Not ideal for high-availability systems
2. Weak consistency: subsequent reads may not see the latest value
3. Eventual consistency: given enough time, all replicas become consistent. Used by Dynamo and Cassandra. Recommended model for this design.

Inconsistency resolution: versioning #

Replication causes inconsistencies among replicas. Versioning treats each modification as a new immutable version.

Conflict example: two servers simultaneously change "name" → "johnSanFrancisco" and "johnNewYork" → conflicting versions v1 and v2.

Vector clocks #

A [server, version] pair associated with a data item. Detects whether versions precede, succeed, or conflict.

Operations:

1. Write to server Si: increment vi if [Si, vi] exists, otherwise create [Si, 1]
2. Version X is ancestor of Y (no conflict) if every participant's counter in Y ≥ corresponding in X
3. Version X is sibling of Y (conflict) if any participant in Y has counter < corresponding in X

Example: D([s0,1],[s1,2]) and D([s0,2],[s1,1]) → conflict.

Downsides: complexity added to the client (must resolve conflicts). Vector clock size can grow: mitigate with a threshold limit (remove oldest pairs). Dynamo paper reports it's not yet a problem in production.

Handling failures #

Failure detection: Gossip protocol #

Better than all-to-all multicasting:

1. Each node maintains a membership list (member IDs + heartbeat counters)
2. Each node periodically increments its heartbeat
3. Each node sends heartbeats to random nodes, which propagate further
4. Membership lists updated on heartbeat receipt
5. If a heartbeat hasn't increased for a predefined period → member considered offline

Handling temporary failures: Sloppy quorum + hinted handoff #

Instead of strict quorum, choose the first W healthy servers for writes, first R healthy servers for reads on the hash ring. Offline servers ignored. When a down server recovers, changes are pushed back (hinted handoff).

Handling permanent failures: Merkle tree (anti-entropy) #

Compare data across replicas, sync only differences.

Merkle tree construction:

1. Divide key space into buckets (e.g. 4 buckets for keys 1–12)
2. Hash each key in a bucket using uniform hashing
3. Create a single hash per bucket
4. Build the tree upward to the root by hashing children

Comparison: start at the root → if hashes match, servers have the same data. If not, traverse children to find unsynchronized buckets → sync only those.

Synchronization amount is proportional to differences, not total data. Typical config: 1M buckets per 1B keys (1,000 keys/bucket).

Handling data center outage #

Replicate data across multiple data centers. Even if one data center is offline, users access data through the other data centers.

System architecture diagram #

Features:

1. Simple APIs: get(key), put(key, value)
2. Coordinator acts as proxy between client and store
3. Nodes distributed on ring via consistent hashing
4. Completely decentralized (add/remove nodes automatically)
5. Data replicated at multiple nodes
6. No single point of failure (every node has the same responsibilities)

Write path (based on Cassandra architecture) #

1. Write request persisted to a commit log file
2. Data saved in memory cache
3. When memory cache is full/reaches threshold → flushed to SSTable on disk (sorted list of <key, value> pairs)

Read path #

1. Check memory cache → if hit, return data
2. If miss, check Bloom filter to identify which SSTables might contain the key
3. SSTables return result
4. Return data to client

Step 4 of 4: Wrap up #

What we designed:

1. Distributed KV store: AP/eventual consistency, consistent hashing, N=3 replication
2. Conflicts resolved via versioning + vector clocks
3. Failures: gossip detection, sloppy quorum + hinted handoff, Merkle tree sync, multi-DC
4. Storage: commit log → memory → SSTable, Bloom-filter read path