📁 topics / 03-key-value-store.md

#Design a Distributed Key-Value Store (DynamoDB Scale)

#1. Problem Statement & Clarifications

A distributed key-value store provides a simple put(key, value) / get(key) interface distributed across multiple machines for scalability, fault tolerance, and low latency. Think DynamoDB, Cassandra, or Riak. The core challenge is balancing consistency, availability, and partition tolerance (CAP theorem) while maintaining sub-millisecond performance.

#Functional Requirements

  1. Put(key, value) — Store a key-value pair (value up to 1MB)
  2. Get(key) — Retrieve value by key with configurable consistency
  3. Delete(key) — Remove a key-value pair
  4. TTL support — Keys can have configurable expiration
  5. Tunable consistency — Client chooses strong or eventual consistency per request

#Non-Functional Requirements

Requirement Target
Read Latency P99 < 5ms (eventual), P99 < 10ms (strong)
Write Latency P99 < 10ms
Throughput 1M+ ops/sec per cluster
Availability 99.99% — prefer AP over CP
Durability No data loss after acknowledged write
Scalability Linear horizontal scaling (add nodes → proportional throughput)

#Out of Scope

#Assumptions

#2. Back-of-Envelope Estimation

#Traffic Estimates

Total ops/sec:          1M (800K reads + 200K writes)
Reads/sec (peak 3x):   2.4M
Writes/sec (peak 3x):  600K
Per-node (100 nodes):   10K ops/sec avg per node

#Storage Estimates

Record size avg:        key (32B) + value (1KB) + metadata (100B) ≈ 1.1KB
New writes/day:         200K/sec × 86400 = 17.3B writes/day
Unique keys (5yr):      ~100B (with overwrites and deletes)
Active dataset:         10B keys × 1.1KB = 11TB
With 3x replication:    33TB across cluster
Per-node (100 nodes):   330GB per node

#Bandwidth

Read bandwidth:         800K × 1.1KB = 880 MB/s cluster-wide
Write bandwidth:        200K × 1.1KB × 3 (replicas) = 660 MB/s
Per-node:               ~15 MB/s (well within NVMe SSD capacity)

#Cache Estimates

Hot keys (top 20%):     2B keys × 1.1KB = 2.2TB
Per-node in-memory:     22GB (feasible with 64GB+ RAM nodes)
Memory-first architecture: keep hot data in RAM, cold on SSD

#3. API Design

#Put (Write)

PUT /api/v1/kv/{key}
Content-Type: application/octet-stream
X-Consistency: quorum              // one | quorum | all
X-TTL: 3600                       // optional, seconds

Body: <raw value bytes>

Response 200:
{
  "key": "user:12345:profile",
  "version": 1716984000123456,     // vector clock / timestamp
  "size": 1024,
  "replicas_ack": 2                // how many replicas acknowledged
}

#Get (Read)

GET /api/v1/kv/{key}
X-Consistency: quorum              // one | quorum | all

Response 200:
Content-Type: application/octet-stream
X-Version: 1716984000123456
X-Replicas-Read: 2

Body: <raw value bytes>

Response 404: { "error": "key_not_found" }

#Delete

DELETE /api/v1/kv/{key}
X-Consistency: quorum

Response 204: No Content

#Batch Operations

POST /api/v1/kv/batch
{
  "operations": [
    { "op": "get", "key": "user:123:profile" },
    { "op": "put", "key": "user:456:session", "value": "base64...", "ttl": 3600 },
    { "op": "delete", "key": "temp:789" }
  ]
}

#4. Data Model

#Storage Engine (LSM-Tree based — per node)

Write path:
  1. Write to WAL (Write-Ahead Log) on disk          → durability
  2. Insert into MemTable (in-memory sorted structure) → fast writes
  3. When MemTable full → flush to SSTable on disk     → persistent
  4. Background compaction merges SSTables             → read optimization

Read path:
  1. Check MemTable (in-memory)                        → fastest
  2. Check Bloom filter for each SSTable               → skip irrelevant files
  3. Binary search in SSTable index                    → find key
  4. Read value from SSTable data block                → return

#Data Record Format

┌──────────┬────────────┬──────────┬───────────┬──────────┬───────┐
│ Key Size │ Value Size │ Key      │ Value     │ Timestamp│ Flags │
│ (4B)     │ (4B)       │ (var)    │ (var)     │ (8B)     │ (1B)  │
└──────────┴────────────┴──────────┴───────────┴──────────┴───────┘
Flags: deleted (tombstone), compressed, TTL-expired

#Partition Map (Consistent Hashing Ring)

Hash ring with virtual nodes:
  - Each physical node → 150-200 virtual nodes on the ring
  - Key → hash(key) → find next N nodes clockwise → those are replicas
  - N = replication factor (typically 3)

Ring metadata stored in coordinator / gossip protocol

#Access Patterns

Query Path Mechanism
Put key Coordinator → N replica nodes (quorum write) Consistent hashing ring lookup
Get key Coordinator → R replica nodes (quorum read) Read from R replicas, return latest version
Delete key Write tombstone marker to N replicas Tombstone garbage collected after grace period
TTL expiry Background compaction removes expired records Checked during compaction + read path

#5. High-Level Design (HLD) & Scale Evolution

#Stage 1: MVP / Single Node (10K ops/sec)

#Stage 2: Growth Scale (100K ops/sec, 10 nodes)

#Stage 3: DynamoDB Scale (1M+ ops/sec, 100+ nodes)

#Architecture Diagram

                    ┌──────────────────────────┐
                    │       Client SDK          │
                    │  (partition-aware routing) │
                    └────────────┬─────────────┘
                                 │
              ┌──────────────────┼──────────────────┐
              ▼                  ▼                  ▼
     ┌──────────────┐   ┌──────────────┐   ┌──────────────┐
     │  Node 1       │   │  Node 2       │   │  Node N       │
     │               │   │               │   │               │
     │ ┌───────────┐ │   │ ┌───────────┐ │   │ ┌───────────┐ │
     │ │Coordinator│ │   │ │Coordinator│ │   │ │Coordinator│ │
     │ └─────┬─────┘ │   │ └─────┬─────┘ │   │ └─────┬─────┘ │
     │       │       │   │       │       │   │       │       │
     │ ┌─────▼─────┐ │   │ ┌─────▼─────┐ │   │ ┌─────▼─────┐ │
     │ │  Storage   │ │   │ │  Storage   │ │   │ │  Storage   │ │
     │ │  Engine    │ │   │ │  Engine    │ │   │ │  Engine    │ │
     │ │ (LSM-Tree) │ │   │ │ (LSM-Tree) │ │   │ │ (LSM-Tree) │ │
     │ │           │ │   │ │           │ │   │ │           │ │
     │ │ MemTable  │ │   │ │ MemTable  │ │   │ │ MemTable  │ │
     │ │ WAL       │ │   │ │ WAL       │ │   │ │ WAL       │ │
     │ │ SSTables  │ │   │ │ SSTables  │ │   │ │ SSTables  │ │
     │ └───────────┘ │   │ └───────────┘ │   │ └───────────┘ │
     └──────────────┘   └──────────────┘   └──────────────┘
              │                  │                  │
              └──────────────────┼──────────────────┘
                                 │
                    ┌────────────▼────────────┐
                    │    Gossip Protocol       │
                    │  (Membership + Failure   │
                    │   Detection)             │
                    └─────────────────────────┘

#Component Breakdown

Component Responsibility Tech Choice
Client SDK Partition-aware routing; retry logic; consistency selection Language-specific SDK
Coordinator Request routing; quorum management; conflict resolution Every node is a coordinator
Storage Engine LSM-tree: WAL + MemTable + SSTables; compaction Custom (RocksDB-like)
Consistent Hashing Ring Key-to-node mapping; virtual nodes; rebalancing In-process ring state
Gossip Protocol Membership; failure detection; ring state propagation Swim / Memberlist
Replication Manager Async/sync replica writes; read repair; hinted handoff Per-node module

#Data Flow

Write Path (Quorum Write W=2, N=3):

Client → Coordinator (any node)
  → 1. hash(key) → find N=3 replica nodes on consistent hash ring
  → 2. Send write request to all 3 replicas in parallel
  → 3. Each replica: WAL append → MemTable insert → ACK
  → 4. Coordinator waits for W=2 ACKs → return success to client
  → 5. Third replica ACKs asynchronously (or via hinted handoff if down)

Read Path (Quorum Read R=2, N=3):

Client → Coordinator
  → 1. hash(key) → find N=3 replica nodes
  → 2. Send read request to all 3 replicas in parallel
  → 3. Wait for R=2 responses
  → 4. Compare versions → return latest value
  → 5. If versions differ → trigger read repair on stale replica

#6. Deep Dive — Core Components

#Consistent Hashing — Detailed Design

Why not modular hashing (key % N)? When adding/removing nodes, modular hashing redistributes almost all keys. Consistent hashing redistributes only K/N keys (K = total keys, N = nodes).

Virtual Nodes:

Physical Node A → VNode_A1, VNode_A2, ..., VNode_A150
Physical Node B → VNode_B1, VNode_B2, ..., VNode_B150

Benefits:
  - Even key distribution (more points on ring)
  - Heterogeneous hardware (stronger node → more vnodes)
  - Smoother rebalancing when nodes join/leave

#Replication & Consistency — Detailed Design

Quorum System: W + R > N guarantees overlap

N = 3 (replication factor)

Strong consistency:  W=2, R=2 (always reads at least one up-to-date replica)
High availability:   W=1, R=1 (faster but may read stale data)
Strongest:           W=3, R=1 or W=1, R=3 (one side sees all replicas)

Conflict Resolution (when W + R ≤ N or partitioned):

Strategy: Last-Write-Wins (LWW) with vector clocks

Vector Clock: [(node_A, 3), (node_B, 1), (node_C, 2)]
  - Each node increments its own counter on write
  - On read: compare vector clocks
    - One dominates → choose it (clear winner)
    - Concurrent (neither dominates) → application resolves OR LWW by timestamp

#Storage Engine (LSM-Tree) — Detailed Design

Write: O(1) amortized
Read:  O(log N) with Bloom filters

MemTable (in-memory, sorted — Red-Black Tree or SkipList)
  └→ flush when > 64MB
      └→ SSTable (immutable sorted file on disk)
          └→ Bloom filter per SSTable (false positive: 1%)
          └→ Sparse index (one entry per 64KB block)

Compaction (background):
  Size-tiered: merge SSTables of similar size → reduce read amplification
  Leveled: maintain sorted levels → better read performance, more write I/O

#Scaling Strategy

Component Strategy
Nodes Add nodes → automatic rebalancing via consistent hash ring
Storage LSM-tree per node; NVMe SSDs; tiered compaction
Hot keys Client-side caching; request routing with read replicas
Replication Configurable N per keyspace; rack/AZ-aware placement
Failure Gossip detection; hinted handoff; Merkle tree anti-entropy

#Caching Strategy

┌──────────────────────────────────────────────────────┐
│ Layer 1: Client SDK Cache                             │
│ • LRU cache for frequently accessed keys              │
│ • TTL-based invalidation; cache size configurable     │
├──────────────────────────────────────────────────────┤
│ Layer 2: MemTable (Per-Node In-Memory)                │
│ • Most recent writes always in memory                 │
│ • 64MB per MemTable; multiple active MemTables        │
├──────────────────────────────────────────────────────┤
│ Layer 3: OS Page Cache                                │
│ • SSTables read via mmap; hot blocks cached by OS     │
│ • Effectively unlimited with enough RAM               │
├──────────────────────────────────────────────────────┤
│ Layer 4: Bloom Filters (Per-SSTable)                  │
│ • Avoid reading SSTables that don't contain the key   │
│ • 10 bits/key → 1% false positive rate                │
└──────────────────────────────────────────────────────┘

#Consistency Models

Data Model Rationale
Writes (W=2, N=3) Strong (within quorum) W+R>N guarantees read sees latest write
Writes (W=1, N=3) Eventually consistent Fast writes; stale reads possible for up to seconds
Membership (gossip) Eventually consistent (< 10s) Ring state converges via gossip protocol
Anti-entropy repair Eventually consistent (< 1h) Merkle tree comparison runs periodically

#7. Low-Level Design — Core Functionality

#Key Algorithms

1. Consistent Hashing with Virtual Nodes

import hashlib
import bisect

class ConsistentHashRing:
    def __init__(self, vnodes_per_node=150):
        self.vnodes = vnodes_per_node
        self.ring = []           # sorted list of (hash, node_id)
        self.ring_hashes = []    # just hashes for bisect
        self.nodes = {}          # node_id → node_info

    def _hash(self, key):
        return int(hashlib.md5(key.encode()).hexdigest(), 16)

    def add_node(self, node_id, node_info):
        self.nodes[node_id] = node_info
        for i in range(self.vnodes):
            vnode_key = f"{node_id}:vnode:{i}"
            h = self._hash(vnode_key)
            idx = bisect.bisect(self.ring_hashes, h)
            self.ring_hashes.insert(idx, h)
            self.ring.insert(idx, (h, node_id))

    def get_nodes(self, key, n=3):
        """Get N unique physical nodes for a key."""
        h = self._hash(key)
        idx = bisect.bisect(self.ring_hashes, h) % len(self.ring)
        result = []
        seen = set()
        while len(result) < n:
            _, node_id = self.ring[idx % len(self.ring)]
            if node_id not in seen:
                result.append(node_id)
                seen.add(node_id)
            idx += 1
        return result

2. Quorum Read/Write

def quorum_write(key, value, n=3, w=2):
    """Write to N replicas, wait for W acknowledgments."""
    nodes = ring.get_nodes(key, n)
    timestamp = time.time_ns()
    version = VectorClock.increment(local_node_id)

    # Send writes in parallel
    futures = []
    for node in nodes:
        f = async_write(node, key, value, timestamp, version)
        futures.append(f)

    # Wait for W acks
    acks = 0
    for f in as_completed(futures):
        if f.result().success:
            acks += 1
        if acks >= w:
            return WriteResult(success=True, version=version)

    # Failed to get quorum — write fails
    raise WriteQuorumNotMet(acks_received=acks, required=w)

def quorum_read(key, n=3, r=2):
    """Read from N replicas, wait for R responses, return latest."""
    nodes = ring.get_nodes(key, n)

    futures = [async_read(node, key) for node in nodes]
    responses = []
    for f in as_completed(futures):
        resp = f.result()
        if resp.found:
            responses.append(resp)
        if len(responses) >= r:
            break

    if not responses:
        raise KeyNotFound(key)

    # Return the latest version
    latest = max(responses, key=lambda r: r.timestamp)

    # Read repair: update stale replicas asynchronously
    for resp in responses:
        if resp.timestamp < latest.timestamp:
            async_repair(resp.node, key, latest.value, latest.version)

    return latest.value

3. Merkle Tree for Anti-Entropy

def build_merkle_tree(key_range, node):
    """Build Merkle tree over a key range for anti-entropy sync."""
    keys = node.get_keys_in_range(key_range)
    leaves = [hash(f"{k}:{node.get_version(k)}") for k in sorted(keys)]

    # Build tree bottom-up
    tree = [leaves]
    while len(tree[-1]) > 1:
        level = tree[-1]
        parent = []
        for i in range(0, len(level), 2):
            left = level[i]
            right = level[i+1] if i+1 < len(level) else left
            parent.append(hash(f"{left}:{right}"))
        tree.append(parent)

    return tree  # root = tree[-1][0]

def anti_entropy_sync(node_a, node_b, key_range):
    """Compare Merkle trees to find divergent keys."""
    tree_a = build_merkle_tree(key_range, node_a)
    tree_b = build_merkle_tree(key_range, node_b)

    if tree_a[-1] == tree_b[-1]:  # roots match
        return  # all keys in sync

    # Traverse tree to find divergent leaves → sync only those keys
    divergent_keys = find_divergent_leaves(tree_a, tree_b)
    for key in divergent_keys:
        sync_key(node_a, node_b, key)  # copy latest version

#Design Patterns Used

Pattern Where Why
Consistent Hashing Key-to-node partitioning Even distribution; minimal redistribution on node changes
Quorum (Sloppy) Read/write consistency Tunable consistency; W+R>N for strong reads
Hinted Handoff Temporary node failure Write to available node; replay when target recovers
Read Repair Stale replica detection Opportunistic consistency fix on every read
Bloom Filters SSTable read optimization Avoid disk reads for keys not in SSTable
Write-Ahead Log Durability before MemTable Crash recovery: replay WAL to restore MemTable

#8. Failure Handling & Edge Cases

#What Happens When X Fails?

Failure Impact Mitigation
Single node down 1/N keys unavailable for strong consistency Hinted handoff; sloppy quorum; read from remaining replicas
Network partition Split brain between node groups AP system: both sides accept writes; resolve conflicts on heal
Disk failure Node's data lost Rebuild from replicas; copy SSTables from 2 healthy replicas
Slow node (straggler) Tail latency spike Read from fastest R replicas; ignore slow one
MemTable crash Recent writes lost Replay from WAL (write-ahead log); no data loss
Compaction stall Read amplification increases Alert on SSTable count; auto-trigger emergency compaction

#Edge Cases

#9. Monitoring & Observability

#Key Metrics

Metric Target Alert Threshold
Read P99 latency < 5ms (eventual) > 20ms
Write P99 latency < 10ms > 50ms
MemTable flush frequency Every 2–5 min > 10/min (too much write pressure)
SSTable count per node < 20 per level > 50 (compaction behind)
Gossip convergence time < 10s > 60s
Replication lag < 1s > 10s
Node disk usage < 80% > 90%
Bloom filter false positive rate < 1% > 5%

#Alerting Strategy

#SLAs / SLOs

Read API (eventual):    99.99% availability, P99 < 5ms
Read API (strong):      99.95% availability, P99 < 10ms
Write API:              99.99% availability, P99 < 10ms
Durability:             99.999999999% (11 nines with 3x replication)
Data consistency:       Convergence within 10s for eventual; immediate for quorum

#10. Trade-off Summary

Decision Chose Over Because
Partitioning Consistent hashing with vnodes Range-based Even distribution; minimal disruption on node add/remove
Storage engine LSM-tree B-tree Optimized for write-heavy; sequential I/O for writes
Default consistency AP (available + partition tolerant) CP Most KV workloads prioritize availability over strict consistency
Conflict resolution Last-Write-Wins (LWW) Vector clocks with app merge Simpler; acceptable for most use cases; vector clocks optional
Failure detection Gossip protocol Centralized heartbeat No single point of failure; scales to 1000+ nodes
Anti-entropy Merkle trees Full data comparison O(log N) comparison; transfer only divergent keys
Replication Sloppy quorum + hinted handoff Strict quorum Higher availability; accept brief inconsistency during failures
Compaction Leveled compaction Size-tiered Better read performance; controlled space amplification

#11. Extensions & Follow-ups

#What Would You Add With More Time?

  1. Secondary Indexes — Allow queries beyond primary key (like DynamoDB GSI)
  2. Multi-Region Replication — Cross-datacenter async replication with conflict resolution
  3. Transactions — Lightweight transactions (compare-and-swap) for conditional writes
  4. Auto-Scaling — Automatic node addition/removal based on load
  5. Tiered Storage — Hot data on SSD, warm on HDD, cold on S3
  6. Change Data Capture (CDC) — Stream mutations to downstream consumers

#How Would This Change at 100x Scale?

#12. Cross-References

Topic Connection
URL Shortener (#1) Uses Redis (a KV store) for URL cache
Rate Limiter (#2) Redis counters are KV store operations
Unique ID Generator (#4) KV store needs unique keys for internal coordination
Distributed Cache (#15) Cache is a specialized volatile KV store

#Building Blocks Used

#Concepts Used