Walkthrough: Distributed Key-Value Store (Dynamo / Cassandra style)

1. Requirements#

Functional

• put(key, value), get(key), delete(key); values are opaque blobs up to ~10 KB-1 MB.
• Tunable consistency per request (fast eventual vs quorum reads/writes).

Non-functional

• Highly available: always writable, even during node failures and network partitions (AP, with tunable C).
• Horizontally scalable to hundreds of nodes; adding/removing nodes moves minimal data.
• Low latency: p99 reads/writes in single-digit ms within a datacenter; durable (replicated, no single disk loss loses data).

2. Estimates#

• Data: 1B keys × (100 B key + 1 KB value) ≈ 1.1 TB raw × replication factor 3 ≈ 3.3 TB; with LSM space amplification (~1.5x) ≈ 5 TB.
• Node sizing: ~1 TB usable per node → 5-6 nodes minimum; plan 10 for headroom and rebalancing.
• Throughput: 100k ops/s cluster-wide; Cassandra-class nodes sustain ~10k-20k ops/s each → 10 nodes fits with quorum amplification (each write hits 3 nodes → 300k replica-writes/s).
• Memory: Bloom filters + index summaries ≈ 1-2% of data per node; memtable 100s of MB.

3. API sketch#

PUT    /v1/kv/{key}   body=value   headers: W=quorum|one|all, If-Match: <version/vector-clock>GET    /v1/kv/{key}                headers: R=quorum|one    -> value(s) + version contextDELETE /v1/kv/{key}               (writes a tombstone)Client library handles: coordinator selection, retries, backoff, version context passing.

4. Data model#

Per node (LSM storage engine):

write -> commit log (append, fsync) -> memtable (sorted in RAM)      -> flush -> immutable SSTable on disk (+ Bloom filter + sparse index)      -> background compaction merges SSTables, drops tombstones after gc_graceread  -> memtable -> row cache -> Bloom filters -> newest SSTables firstKey metadata: version (vector clock or last-write timestamp), tombstone flag, TTL

Cluster metadata (membership, token ranges) gossiped; or held in Zookeeper/etcd in simpler designs.

5. High-level design#

client -> [coordinator node (any node)] --hash(key)--> position on consistent-hash ring            |  preference list: next N=3 distinct physical nodes clockwise (vnodes for balance)            |-> replica A  (write to commit log + memtable)            |-> replica B            |-> replica C            <- ack after W replicas respond (W=2); read waits for R=2, returns newest, repairs stale (read repair)   ring: [A][B][C][D][E][F] ... each physical node owns ~128-256 virtual tokens  membership + failure detection: gossip protocol (every node exchanges state with a few peers each second)  temporary failure: sloppy quorum + hinted handoff (D stores hint for C, replays when C returns)  permanent failure / drift: anti-entropy via Merkle trees between replicas

6. Deep dives#

Partitioning and replication

• Consistent hashing with virtual nodes: adding a node moves ~1/N of keys; vnodes smooth load and let heterogeneous nodes take proportional ranges. Replicate to the next N distinct physical nodes (rack/AZ-aware so one AZ failure doesn't take all replicas).

Consistency: quorums and conflict resolution

• N=3, W=2, R=2 → R+W>N guarantees overlap: the newest write is seen by every quorum read in the common case. W=1 for fast fire-and-forget; R=1 for cheap stale reads; W=3 for strong durability.
• Concurrent writes to the same key on different replicas → siblings. Detect with vector clocks ([(A,1),(B,2)]), resolve by last-write-wins (Cassandra, needs synced clocks) or return siblings to the client to merge (Dynamo, Riak) or use CRDTs for counters/sets.
• Sloppy quorum: during partition, write to the next healthy node outside the preference list and store a hint; keeps availability but weakens R+W>N until handoff completes.

Failure handling

• Failure detection: gossip with phi-accrual detector (suspicion level rather than binary timeout) to avoid flapping.
• Temporary: hinted handoff. Permanent: operator removes node; ranges reassigned; anti-entropy with Merkle trees compares replica hash trees and syncs only differing ranges.
• Deletes are tombstones so a resurrected replica can't bring a key back; purge after gc_grace_seconds (Cassandra default 10 days) once repair has run.

Storage engine

• LSM tree: sequential writes (fast ingest), Bloom filters cut read amplification (a miss costs ~0 disk reads), compaction (size-tiered vs leveled) trades write amplification vs read/space amplification.

7. Bottlenecks & scaling#

• Hot keys/partitions: consistent hashing balances nodes, not keys; mitigate with key salting (key#0..9), client-side caching, or replicating hot keys to more nodes.
• Large values: cap value size (DynamoDB 400 KB, Cassandra recommended < 1 MB); store blobs in object storage and keep pointers.
• Compaction storms: throttle compaction IO; leveled compaction for read-heavy, size-tiered for write-heavy.
• Tombstone buildup: avoid delete-heavy queue-like workloads; TTL instead of explicit deletes where possible.
• Coordinator overhead: clients can be token-aware (route directly to a replica) to save a hop.
• Multi-DC: replicate per DC with local quorums (LOCAL_QUORUM) to avoid WAN latency on every write; cross-DC async.
• Operations: rolling restarts, repair scheduling (run full repair within gc_grace), monitoring of pending hints, dropped mutations, p99 latency per node.

8. Words to say#

consistent hashing, virtual nodes, preference list, replication factor N=3, quorum R+W>N, tunable consistency (ONE/QUORUM/ALL/LOCAL_QUORUM), sloppy quorum, hinted handoff, read repair, anti-entropy, Merkle tree, gossip protocol, phi-accrual failure detector, vector clock, last-write-wins, siblings/CRDT, tombstone + gc_grace, LSM tree, commit log, memtable, SSTable, Bloom filter, compaction (size-tiered vs leveled), write/read/space amplification, rack-aware replica placement, token-aware client, hot key salting.