How to design a distributed key-value store

A distributed key-value store (Dynamo/Cassandra style) keeps N copies of each key and turns on one decision: when the copies disagree, who notices and who resolves it.

Interface and placement

• get(key) -> (siblings, context): returns a list of values, not one; two values kept side by side are siblings. A caller that can’t handle len>1 has a latent bug.
• put(key, value, context): context is an opaque token from a prior get; empty context = “I accept this becomes a sibling.” delete writes a tombstone, not an erase.
• Consistent hashing picks owners: hash key to a ring, walk clockwise, first N distinct nodes = the preference list. N = replication factor = 3.
• No range scans, no secondary indexes, no multi-key transactions. Giving these up pays for everything else.
• Any node is coordinator (whichever the client hit); the ring is a pure function of membership, so no master or config tier.

Quorum: W + R > N

• W = replicas that must ack a write; R = replicas that must answer a read. Overlap guaranteed by pigeonhole: two subsets share W + R - N members; at N=3, W=R=2 that is 1, so the read set can’t miss the last completed write.
• Coordinator sends to all N, returns on first W/R. R=1 = fastest of three (a free hedged request), not one random node.
• Default: W=2, R=2. W=3 is ~1,000x less available for no gain; (1,1) gives up overlap and returns stale with no error during any incident.
• Quorum does not give linearizability (concurrent writes, in-flight writes, sloppy quorum all break it). Read-your-writes needs one coordinator; monotonic reads need read repair; true linearizability needs consensus (Raft/Paxos).

Vector clocks: detect, never resolve

• Vector clock = map node -> counter, carried per version; coordinator increments its own entry on each accepted write.
• A descends B if A[n] >= B[n] for all n in B. Neither descends -> concurrent -> keep both as siblings. Merged context = pointwise maximum of sibling clocks.
• Record = 1,088 B: key 32, value 1,024, timestamp 8, clock entry 24 (node, counter, timestamp). One entry in the common case.
• Gotchas: dedup on (clock, value) not clock alone (blind writes share a clock); read-repair every replica that answered, not just first R; never treat empty clock as ancestor.
• Clock grows one entry per distinct coordinator; Dynamo’s 10-entry cap + evict-oldest can silently drop a concurrent write. Fixes: server-side version vectors (bounded by N), dotted version vectors (kill blind-write siblings), or LWW (Cassandra default, discards one write by wall clock; NTP skew ~10 ms, wrong for mutable data).

Merkle trees: one bad key in a million

• Binary hash tree: leaf = hash of one (key, version), internal node = hash of children; root is a 32 B fingerprint of the whole range. Used for anti-entropy repair.
• Matching roots prove 2^20 = 1.05 M keys identical in 32 bytes. On mismatch, descend: root + 2 hashes/level = 41 hashes (O(log n)); practical scheme ships top 11 levels then one subtree = 2 round trips, 131 KB, 8,708x less than streaming (1.14 GB).
• Building is O(n) and a full range read (2.04 TB = 34 min/node) -> repair is a weekly scheduled job, not continuous.
• Assumes identical range boundaries (serialize topology changes) and few differing keys; a node down a day differs everywhere, where streaming would have been cheaper.

Storage engine + bloom filter

• Write path: append WAL + fsync -> insert memtable (sorted RAM) -> ack. Background: flush to immutable 64 MB SSTable, then compaction. One sequential append, no in-place update.
• Size-tiered compaction (~4.5x write amp, 92 MB/s) beats leveled (write amp 1+1+T*L = 52, ~1,061 MB/s > 1 GB/s NVMe) here because the workload is write-dominated; pay in read amp + ~2x space.
• Reads check memtable then SSTables newest-first; size-tiered leaves ~8 overlapping files, 7/8 guaranteed misses.
• Bloom filter (per SSTable, in RAM): hash key to k bits; “no” = proof of absence (skip disk), “maybe” = ≤1 wasted read. No false negatives. FP rate = 0.6185^(m/n); every +4.8 bits/key divides FP by 10.
• At 10 bits/key (k=7): 0.82% FP, 2.34 GB RAM/node. Cuts 8 disk reads to 8 x 0.0082 = 0.066 (122x); without filters ~6 devices/node to serve reads, with them <1. Not an optimization — it makes the LSM read path viable.
• Limits: exact keys only (no ranges), can’t forget (FP drifts up as SSTable ages), must stay resident in RAM or the saving inverts.

Hinted handoff, sloppy quorum, membership

• Sloppy quorum: if a preference node is down, coordinator writes to the next healthy node, which stores a hint and replays it on return (hinted handoff). The moment it fires, W+R>N stops holding — writer and reader sets can be disjoint.
• Two rates 1,000x apart at 99.9%/node: 1 of 3 down ≈ 0.30% -> ~25.9 M writes/day route outside the list (sizes hint volume); 2 of 3 down ≈ 0.0003% -> ~26,000 writes/day exposed to a stale read (the correctness number). Alert on both separately.
• Hint storm: 10-min outage = 11.25 M hints / 12.24 GB; 3-hr window (Cassandra default) = 220 GB. Mitigate by rate-limiting replay and rejoining for writes before reads. Past the window, only Merkle repair converges.
• Gossip (every 1 s, random peers) spreads membership in O(log S) rounds. Phi-accrual detector: phi = -log10 P(next heartbeat later than t); phi=8 ≈ one-in-100M silence, ~18.4 s at 1 s mean, threshold widens with the network. “Down” never auto-removes from the ring — decommission is human-gated (removing a node moves 2.04 TB).

CAP / PACELC for this store

• CAP describes only behavior during a partition; P isn’t a choice. A single node isn’t “CA.” PACELC: during Partition trade A vs C; Else trade Latency vs Consistency (the trade you pay every request).
• Here W is the per-request knob: W=2 = PC (minority refuses), W=1 = PA (both sides accept, divergence = siblings), W=3 = unavailable everywhere.
• W=1 sibling cost: a 60 s / 100k-writes partition yields ~900 conflicting keys of 6 M writes (0.015%), matching Dynamo’s 99.94% single-version — but this assumes uniform writes and fails under a Zipf hot set.

Numbers to keep

• 10 B objects, 1,088 B each -> 10.88 TB logical, 32.64 TB at RF 3, 2.04 TB and 1.875 B keys per node (S=16).
• Steady writes = 16% of a 1 Gbps NIC; the other 84% is the whole budget for repair, hint replay, and rebalance — which all fire at once in an incident.
• Tombstone resurrection risk: gc_grace (10 days) must exceed the repair interval, not its duration; weekly vs 10 days is 1.4x margin — a slipped week, not a slow run, is the danger.
• The one line: detect disagreement precisely and hand it back, never guess; W=2, R=2 at N=3 is the default that makes it work.