38. Distributed Key-Value Store (Dynamo)

Business context. An internal platform store that many teams build on. The success metric is adoption plus SLO attainment. It must offer predictable latency at any scale and tunable consistency, and stay highly available for writes (the shopping-cart lesson: never reject an add-to-cart).

Requirements. get(key), put(key, value), optional conditional put; values up to 1 MB; millions of ops/s; p99 < 10 ms; 99.99% availability; horizontal scaling with no downtime; survive an AZ loss.

Design (Dynamo/Cassandra lineage).

Client ─▶ any node (coordinator) or a partition-aware client
  Partitioning: consistent hashing with vnodes; replication factor N=3 across distinct AZs
  Write: coordinator sends to N replicas, waits for W acks
  Read:  coordinator queries R replicas, returns newest; read-repairs stale replicas
  Per node: commit log (WAL) → memtable → SSTables (LSM) + bloom filters + compaction (§6.3)
  Membership & failure detection: gossip + phi-accrual failure detector
  Temporary failure: sloppy quorum + hinted handoff
  Permanent divergence: anti-entropy with Merkle trees (compare tree hashes, sync only differing ranges)
  Conflicts: vector clocks + app merge, or LWW (simpler, loses concurrent writes)

Deep dives.

Tunable consistency: per request, R/W = ONE / QUORUM / ALL. Explain what QUORUM/QUORUM guarantees (§10.2) and what it doesn't.
Adding a node: it takes over vnodes from many existing nodes, and data streams in. Throttle the streaming to protect foreground latency.
Hot partitions: consistent hashing doesn't fix hot keys. Add per-key rate limits, caching, or client-side key splitting (§9.3).
Tombstones and deletes: deletes are writes. Tombstones must outlive the repair window (gc_grace), or deleted data comes back to life.
Compaction strategy: size-tiered (write-optimized) vs leveled (read-optimized, more write amplification) vs time-window (TTL'd time series).

Alternative design to mention: single-leader per range with Raft groups (the TiKV/CockroachDB/Spanner lineage) gives strong consistency and transactions, at the cost of leader-based write latency and availability during leader elections. The choice depends on what the platform's customers need. Many modern platforms offer strong consistency by default, because application teams handle eventual consistency badly.

Strategic lens. As a platform: per-tenant quotas, noisy-neighbor isolation, self-serve provisioning, capacity forecasting, and clear contracts (documented consistency semantics, limits, error codes). Operational excellence decides adoption more than the internals do.

🏗️ Flow: a node joins the ring.

 Ring with vnodes: A(256 tokens) B(256) C(256). New node D joins with 256 tokens.
   1. D gossips "joining"; computes ranges it will own (≈1/4 of ring, taken from A, B, C in small slices)
   2. Stream data for those ranges from current replicas (throttled, e.g. 50 MB/s)
   3. During streaming: writes for those ranges go to old owners AND D (pending ranges)
   4. Streaming complete → D "normal"; ring metadata updated via gossip
   5. Old owners run cleanup (delete data they no longer own)
 Failure during join → resume streaming; never mark D normal until all ranges complete

🌍 How it's built in production.

Amazon DynamoDB is not Dynamo. The 2022 USENIX ATC paper describes DynamoDB's real design: partitions replicated with Multi-Paxos and a leader per partition, with strongly consistent reads available. Its lineage is Dynamo's ideas plus the operational lessons of a managed service (admission control, adaptive capacity, global admission control). Mixing up the two is a common interview mistake.
Apple and Netflix run very large Cassandra fleets. Netflix built tooling (Priam) for backup, token management, and autoscaling.
Discord (§6.7) shows Cassandra's operational pain points: GC pauses, compaction backlogs, and hot partitions from busy channels.
FoundationDB (Apple): an ordered, transactional KV store famous for deterministic simulation testing, where the whole cluster runs in a simulator that injects faults reproducibly. The bar for correctness testing in distributed storage.

⚠️ Corner cases.

Hinted handoff overflow: a node is down for hours, and the hints fill the disks of the nodes storing them. Cap hint windows, and after that rely on repair.
Repair not run → resurrection: deleted data returns after gc_grace_seconds (§10.8). Run repairs on a schedule shorter than gc_grace.
Large partitions: one partition key with 10 GB of rows causes slow reads, compaction pain, and heap pressure. Bucket the key (time buckets, as Discord does).
LWW with clock skew: a node with a fast clock "wins" every conflict. Run NTP monitoring and prefer server-assigned timestamps from one coordinator, or use version vectors.
W=1 + R=1 for speed: the user writes, then reads from another replica, and their data is "gone". Use quorum writes and reads for read-your-writes.
Rebalancing during a failure: adding nodes while one is down concentrates streaming load on fewer replicas. Fix the failure first, then scale.