Design a Distributed Key-Value Store

A distributed key-value store is a fundamental building block of modern distributed systems. Companies like Amazon (DynamoDB), Google (Bigtable), and Facebook use them to store user sessions, shopping carts, product catalogs, and more. This is a classic system design question that tests your understanding of distributed systems fundamentals.

We'll walk through this design using the Interview Framework to structure our approach.

Phase 1: Requirements

Functional Requirements

• put(key, value): Store a value associated with a key
• get(key): Retrieve the value for a given key
• delete(key): Remove a key-value pair (optional, can be implemented as put with tombstone)
• Configurable consistency: Allow applications to tune consistency via quorums (quorum consistency; still eventual under partitions)

Non-Functional Requirements

• High availability: The system should always be writable (AP in CAP theorem)
• Scalability: Handle millions of keys across thousands of nodes globally
• Fault tolerance: Continue operating despite node failures
• Low latency: Single-digit millisecond reads and writes
• Tunable consistency: Support configurable read/write quorums

Capacity Estimation

Let's design for Amazon-scale:

• 100 million DAU, each performing ~100 operations/day
• 10 billion operations/day → ~115K QPS average, ~350K QPS peak
• 1 billion keys, average value size 1 KB
• Storage: 1B × 1KB = 1 TB (with 3x replication = 3 TB)

Phase 2: Data Model

Core Entities

KeyValue:
  key: string (hashed to 128-bit identifier via MD5 or another fixed-length hash)
  value: blob (typically < 1MB, larger objects go to blob store)
  version: VectorClock
  timestamp: int64 (for TTL/compaction, not for ordering updates)
  checksum: string (for integrity verification)

Vector Clock for Versioning

VectorClock:
  entries: map<node_id, counter>
  # Example: {A: 2, B: 1, C: 1}

Vector clocks track causality between updates. If clock A has all counters ≥ clock B's counters, then A happened after B. Otherwise, they're concurrent (conflict).

Phase 3: API Design

Protocol Choice: Custom Binary Protocol

For internal service-to-service communication, we use a custom binary protocol over TCP for efficiency. For external access, we provide a REST API.

Internal API

# Write operation
put(key: bytes, value: bytes, context: VectorClock) -> VectorClock

# Read operation
get(key: bytes) -> (value: bytes, context: VectorClock)
# May return multiple values if conflicts exist

# Delete operation
delete(key: bytes, context: VectorClock) -> VectorClock

External REST API

PUT /kv/{key}
  Body: { "value": "base64_encoded_data" }
  Response: { "version": "clock_token" }

GET /kv/{key}
  Response: {
    "value": "base64_encoded_data",
    "version": "clock_token",
    "conflicts": [...] // if multiple versions exist
  }

Phase 4: High-Level Design

Architecture Overview

Coordinators are a logical role, not a dedicated tier. Any storage node can act as a coordinator; the load balancer or client library simply picks one.

Key Components

1. Client Library: Routes requests directly to appropriate coordinator (partition-aware) for lower latency
2. Coordinator Node: Any storage node can handle read/write operations and manage quorum
3. Storage Nodes: Store key-value pairs, organized in a consistent hash ring
4. Gossip Protocol: Propagates membership and failure information

Consistent Hashing

We partition data across nodes using consistent hashing:

1. Hash each key to a point on a ring (0 to 2^128 - 1)
2. Hash each node ID to multiple points (virtual nodes)
3. Key is assigned to the first node found clockwise from its hash position

Why virtual nodes?

• Load balancing: Each physical node has multiple positions, distributing keys more evenly
• Heterogeneous hardware: Powerful nodes can have more virtual nodes
• Graceful scaling: Adding/removing nodes only affects neighboring positions

Data Flow: Write Path

Data Flow: Read Path

Storage Engine (Per Node)

Each node uses an LSM-tree based storage engine:

Phase 5: Scaling & Deep Dives

Configurable Consistency (Quorum)

The system uses three parameters for tunable consistency:

Constraint: r + w > n ensures overlap between read and write sets, increasing the chance you read the latest write (concurrent writes can still yield conflicts).

Handling Temporary Failures: Sloppy Quorum & Hinted Handoff

When a node is temporarily unavailable:

1. Sloppy quorum: Route requests to next healthy node in the ring
2. Hinted handoff: The temporary node stores data with a "hint" indicating the intended recipient
3. When the original node recovers, hints are replayed to bring it up to date

Handling Permanent Failures: Merkle Trees

For permanent node failures or data corruption, we use Merkle trees for efficient anti-entropy:

1. Each node maintains a Merkle tree per key range
2. Leaf nodes = hash of individual keys
3. Parent nodes = hash of children
4. Nodes compare root hashes to detect differences
5. Drill down to find exactly which keys differ

Benefit: Compare millions of keys by exchanging just a few hashes. Only transfer keys that actually differ.

Conflict Resolution with Vector Clocks

Vector clocks track causality across distributed updates:

Initial: {} (empty)

Event E1 on Node A: {A:1}
Event E2 on Node A: {A:2}

Network partition occurs...

Event E3 on Node B (based on E2): {A:2, B:1}
Event E4 on Node C (based on E2): {A:2, C:1}

Partition heals...

E3 and E4 are CONCURRENT (neither happened-before the other)
→ Conflict! Return both values to client for resolution.

Client merges and writes E5: {A:3, B:1, C:1}

Failure Detection: Gossip Protocol

Nodes use gossip protocol for membership and failure detection:

1. Each node maintains a membership list with heartbeat counters
2. Periodically, each node picks random peers and exchanges membership info
3. If a node's heartbeat hasn't increased for a threshold time, mark it as suspected
4. After longer timeout, mark as failed and trigger data redistribution

Advantages:

• Decentralized, no single point of failure
• Scalable: O(log N) rounds to propagate information
• Bandwidth efficient: only exchange deltas

Read Repair

Opportunistic consistency fix during reads:

1. Coordinator reads from r nodes
2. If versions differ and are comparable, return latest to client
3. If versions are concurrent, return siblings and let client merge
4. If a single latest exists, asynchronously repair stale replicas; otherwise the client merge write becomes the repair

This gradually heals inconsistencies without dedicated background processes.

Common Pitfalls

Interview Checklist

Before concluding, verify you've covered:

• Clarified consistency requirements (quorum behavior and conflict handling)
• Explained consistent hashing with virtual nodes
• Described quorum-based reads/writes with tunable n, r, w
• Covered failure handling: sloppy quorum, hinted handoff
• Explained conflict resolution: vector clocks, client-side merge
• Mentioned anti-entropy: Merkle trees for detecting/repairing drift
• Described gossip protocol for failure detection
• Touched on storage engine (LSM-tree) if time permits

Summary

This design mirrors Amazon's Dynamo, which influenced many modern systems including Cassandra, Riak, and Voldemort. The key insight is trading consistency for availability—always accepting writes and resolving conflicts later—which makes it ideal for use cases like shopping carts where availability is paramount.