shreyansh

Distributed Key-Value Store

TL;DR — 30-Second Glance

  • Dynamo-style AP store: every node can serve reads/writes; no leader, no failover — consistent hashing with vnodes spreads keys and smooths membership changes.
  • Tunable sloppy quorum (N/R/W, R + W > N) trades latency for consistency per operation; partial failures still make progress via hinted handoff.
  • Concurrent writes keep siblings versioned by vector clocks; conflicts resolve on read (read-repair) or merge (app/CRDT).
  • Merkle-tree anti-entropy reconciles replicas in the background without shipping all data; gossip spreads membership.
  • This note links the Theory building blocks rather than re-deriving them — read it as the assembly manual.

1. Requirements

Functional

  • GET / PUT / DELETE by key; opaque values; conditional write via version context.
  • Tunable consistency per request (e.g. strong-ish reads for auth tokens, fast writes for carts).
  • List siblings on conflict; delete = tombstone with GC.
  • Membership changes (add/remove nodes) with minimal data movement.
  • Multi-DC capable (replicas per DC; local-quorum options).

Non-functional

  • Availability first: writes succeed during partitions/node loss (shopping-cart heritage).
  • Latency: p99 single-digit ms in-region; quorum waits bounded by timeout + hedge.
  • Scale: 10 TB+ data, 100 k+ ops/s, linear scale-out by adding nodes.
  • Durability: acked writes survive N−1…N−k failures per quorum config.

Out of scope

  • Secondary indexes, range scans, transactions (use a CP store instead); re-deriving quorum theory (see Related).

2. Capacity Estimation

Design target example (state assumptions in the interview, then derive):

Quantity Assumption
Data 10 TB unique bytes, avg value 1 KB ⇒ ~10 B keys
Throughput 100 k reads/s, 50 k writes/s
Per-node budget 1 TB disk, 10 k ops/s (with headroom)
Replication N = 3

Derivations:

  • Nodes for data: 10 TB × 3 / 1 TB ≈ 30 nodes; for throughput: 150 k / 10 k = 15 nodes ⇒ data-bound at 30 nodes (+20% headroom ⇒ ~36). Quote both bounds — interviewers check that you size on the max.
  • Quorum latency: W=2 of N=3 needs the 2nd-fastest replica; with per-node p99 3 ms, quorum p99 ≈ ~5 ms + 1 network hop. R=1 cuts read latency ~40% but risks stale reads (window = replication lag + hint backlog).
  • Handoff backlog: one dead node with 1/36 of writes (1.4 k/s × 1 KB) accrues hints at ~1.4 MB/s — trivially replayable; backlog alerts fire if hints age past ~1 h.
  • Merkle sync cost: per-partition trees of ~4 KB ⇒ full-ring comparison ≈ nodes × partitions × 4 KB ≈ tens of MB to find which ranges diverge, then only divergent ranges ship (see §6C).

3. API Design

PUT /v1/keys/{key}
{ "value": "<base64>", "context": "<opaque vector clock; empty = new>" }
-> 200 { "context": "<new clock>" } | 409 { "siblings": [...] } (if configured)

GET /v1/keys/{key}?quorum=r2 | r1 | all
-> 200 { "values": [{ "value", "context" }],   # >1 entry = siblings
         "repaired": true }
GET /v1/keys/{key}?siblings=true   # explicit conflict inspection

DELETE /v1/keys/{key} { "context": "…" } -> 200  # tombstone, GC after grace

# Ops
GET /v1/cluster/membership -> { nodes[], tokens[], status }
POST /v1/cluster/join { node_id } | POST /v1/cluster/leave { node_id }

The context is an opaque vector clock the client echoes back — the server never asks clients to understand it, only to return it (causality without client logic). See Version vectors and Leaderless replication.

4. Data Model & Storage

Layer Design
Partitioning Consistent hashing ring, ~256 vnodes/node; key → coordinator = first N distinct physical nodes clockwise. Vnodes smooth load when nodes differ in size and shrink moved ranges on membership change.
Per-node engine LSM (memtable + SSTables), keyed by (token, key); value = {bytes, vector_clock, ts}; deletes = tombstones with GC grace > max hint age.
Replication Preference list of N; coordinator writes all N (or fastest N healthy + hints for the rest); reads from R with digest mismatch triggering full-fetch + read-repair.
Hint store Local hints:{target_node} → [writes] with TTL; replayed on gossip-detected recovery, then deleted.
Merkle state One tree per vnode-range; leaves = hashes of key-range buckets; roots gossiped/exchanged pairwise.

Prerequisite reads (not duplicated here): Quorums, Partitioning, Rebalancing.

5. High-Level Design

Flow: client (or router) → coordinator node (any node, via ring-aware driver) → parallel N replica RPCs → quorum check → respond; background: gossip (membership), hint replay, Merkle anti-entropy, read-repair on the read path.

loading diagram…

No component is special: coordinators are stateless per request, replicas symmetric. A load balancer just needs to reach any healthy node; smarter drivers route to a replica to save a hop.

6. Deep Dives

A. Consistent hashing, vnodes, and sloppy quorum

  • Ring + vnodes: each physical node owns ~256 virtual tokens scattered on the ring. Adding a node steals thin slices from many owners (not one big range) ⇒ balanced, incremental transfer; removing one reverses it. Heterogeneous hardware = proportional vnode counts.
  • Sloppy quorum: the preference list may substitute a healthy node for a dead one (first N healthy clockwise), and the substitute holds a hint for the dead owner. This is what keeps W available during failures — at the cost of temporary placement skew that anti-entropy later heals.
  • Tuning table (N=3): W=1/R=3 (fast writes, slow consistent reads); W=2/R=2 (balanced, R+W>N ⇒ read-your-quorum); W=3/R=1 (durable writes, maybe-stale fast reads); W=1/R=1 (max speed, eventual everything — carts/session state). Timeouts + hedged requests bound tail latency: after p95, send a 4th speculative replica read and take the first R.
  • Failure math to quote: with W=2, a write survives any single replica loss post-ack; with sloppy hints it succeeds through a loss. Durability window = time until hints replay + Merkle converges.

B. Vector clocks, siblings, and read-repair

  • Each version carries a vector clock (node:counter pairs, pruned by last-write-wins metadata and size caps). On write, the coordinator increments its entry and merges the client's context — capturing causality, not wall-clock order (clocks lie; counters don't — see Lamport clocks).
  • Concurrent writes ⇒ siblings (both kept, both returned). Resolution paths: (1) last-write-wins by timestamp (simple, lossy — fine for sessions); (2) client merge on next write (carts: union the items); (3) CRDT values where the merge is associative/commuting — see CRDTs.
  • Read-repair: a read that finds digest disagreement pushes the newest version to laggards inline (bounded: repair asynchronously past a threshold so reads stay fast). Coupled with Merkle sweeps, the system converges without any read ever blocking on full sync.

C. Hinted handoff and Merkle anti-entropy, end to end

  • Write with one replica down (N=3, W=2): coordinator writes A, B, and hands C's copy to D as a hint (for: C, until: T). Client gets ack (W=2 met). Gossip marks C alive ⇒ D streams hints to C, C applies (newer-clock wins per key), hints deleted. If C stays dead past hint TTL, hints expire and Merkle catches the gap later — the two mechanisms overlap deliberately.
  • Merkle exchange: two replicas compare tree roots for a shared range; equal ⇒ done (one hash compared, zero bytes shipped); unequal ⇒ descend to differing leaves and exchange only those key buckets. Cost scales with divergence, not data size — this is the entire point, and the sentence to say in interviews.
  • Tombstone discipline: deletes replicate like writes; GC grace period must exceed max hint age + Merkle interval, else a resurrected replica reintroduces deleted keys (the classic "zombie write"). Quote: grace = 7 days is typical; shrinking it is a durability decision, not a cleanup tweak.

7. Bottlenecks, Trade-offs & Monitoring

Approach Pros Cons Use when
N=3, R=2, W=2 Balanced, quorum overlap 2-replica latency on every op Default; adjust per workload
Sloppy quorum + hints Available through failures Placement skew, hint ops Always in AP mode; strict lists only for placement-sensitive data
Siblings + app merge Never lose a write Client complexity Carts/collab; LWW for cache-like data; CRDTs where mergeable
Merkle anti-entropy Converges with minimal bytes Background I/O, zombie-write edge Always on; interval tuned by churn

Bottlenecks: hot keys (quorum can't fix a celebrity key — cache in front, split, or accept it); coordinator CPU on wide preference lists; hint-disk exhaustion during long outages (cap + alert, then shed oldest with metrics); Merkle I/O competing with foreground (throttle + off-peak full sweeps).

Monitoring: per-op quorum latency (R/W separately), hint count/age per target, Merkle divergence rate, sibling rate (conflict health), read-repair rate, gossip convergence time, tombstone age vs GC grace, vnode balance (stddev of owned keys).

Interview one-liners

  • Q: Why vnodes instead of one token per node? A: Balanced slices, proportional capacity, incremental movement on membership change.
  • Q: R + W > N — why? A: Quorum overlap guarantees at least one replica saw the latest acked write (see Quorums).
  • Q: Node dies mid-write — lost? A: No: sloppy quorum + hinted handoff; ack on W healthy, replay on recovery, Merkle as backstop.
  • Q: Deletes coming back from the dead? A: Tombstones with GC grace > hint age + Merkle interval.