Design a key-value store
Design a distributed key-value store like Dynamo. Start with one node holding a map and a log, grow it with an LSM tree, a hash ring of about 70 nodes, quorums and repair, then stress it with what quorums promise, clashing writes and adding nodes, with code.
“Design a distributed key-value store. Highly available, scales horizontally. Think Dynamo.”
The interface is just two calls: get(key) and put(key, value). Everything hard sits underneath
them. How does one machine store billions of keys safely? How are the keys split across 70
machines? What happens when a copy misses a write? And what can a client really rely on when it
reads?
That last question is where most answers go wrong. So this page starts with one node and lets the numbers grow it, up to copies that repair themselves. Then it tests the result: what quorums promise, what happens when two writes clash, and how to add nodes safely.
The answer in one minute
Section titled “The answer in one minute”If you only have a minute, this is the design. The rest of the page explains each part.
- Each node stores data in an LSM tree. Writes go to a log on disk and a sorted table in memory. When that table fills up, it is written to disk as a sorted file. Writing is always fast, because the disk only ever appends.
- Keys are spread over about 70 nodes with consistent hashing. Each key’s hash lands on a ring, and the next nodes clockwise own it. Adding a node moves only a small slice of the data.
- Every key has 3 copies, in 3 different racks. A write goes to all 3 and succeeds after 2 answer. A read asks 2. Because 2 + 2 is more than 3, a read always reaches at least one copy that has the latest successful write.
- There is no leader. Any copy can take a write. So writes keep working when a node is down. The price is that copies can disagree, and two writes can clash.
- Three repair tools bring copies back in sync: notes saved for a node that was down, fixes made during reads, and a regular comparison of all data.
Scope it
Section titled “Scope it”A few questions change this design a lot. The answers are the assumptions the rest of the page builds on.
Design a distributed key-value store. It should be highly available and scale horizontally.
Whether you pin down the consistency callers need, the data size and the read to write mix before choosing anything, because those decide the replication scheme. Three minutes. Naming Cassandra in the first minute, or asking which language to use, reads badly.
These four questions matter most:
- How much data? 20 billion keys, with values of about 1 KB, at most 1 MB.
- How busy? 200,000 reads and 100,000 writes a second, three times that at peak. A third of the traffic is writes, so writes must be cheap.
- Must a write succeed when copies are down? Yes. Things like shopping carts and sessions must always be writable. If the answer were “never read old data”, we would need a leader for each group of keys, and a very different design.
- Range scans? No, only single keys. Scans would change how keys are split.
From these answers we make four promises:
- Always writable. 99.99% of the time, a write succeeds if 2 of a key’s 3 copies are reachable.
- Fast. Under 10 ms for 99% of requests, and under 3 ms for half of them.
- Durable. A write is on disk on at least 2 machines, in different racks, before we say “done”.
- Honest consistency. Copies agree within seconds after a failure heals. And we’ll say exactly what a read can and can’t promise.
We leave out transactions across keys, secondary indexes and query languages.
Go deeperMore questions, and what each changes1 min
| Question | We assume | If the answer were different |
|---|---|---|
| Does anyone need compare-and-set? | A few callers, later | If most do, every write needs consensus: a different store |
| One datacenter or several? | Two, both serving traffic | One halves the hardware and removes clashes between datacenters |
| Values of 10 KB? | No, about 1 KB | Disk space, not traffic, would decide the number of nodes |
| Read-mostly? | No, a third writes | Read-mostly data would favour a B-tree engine or a cache |
The caller picks, per request, how many copies must answer: ONE, QUORUM (2 of 3), or ALL. And each table picks how clashing writes are handled: the latest timestamp wins, or both versions are kept for the application to merge.
The numbers that decide it
Section titled “The numbers that decide it”Now we work out how big this is. The key is to count operations on copies, not client requests.
- Keys stored
- 20 billion
- Average value size
- 1 KB
- Reads a second, peak
- 600,000
- Reads per write
- 2 to 1
- Replication factor
- 3×
- Replica ops per node a second
- 50,000
- Data per node
- 2 TB
20 billion keys × about 1.1 KB is 22 TB. Three copies make 66 TB. The LSM engine keeps about 30% extra that cleanup hasn’t removed yet: about 86 TB on disk.
So we data lives on disk; it will never fit in memory.
At peak, 600,000 reads a second each touch 2 copies, and 300,000 writes a second each touch 3. That is 2.1 million operations a second.
So we size the cluster for operations on copies, which are seven times the client writes.
A 16-core node with fast SSDs handles about 50,000 operations a second. At 60% busy, that’s 70 nodes. Disk space alone would need only 43.
So we buy CPU cores and fast disks, not bigger disks. Each node holds about 1.2 TB.
A dead node’s 1.2 TB copies back from the others in 2 to 3 hours, at about 150 MB a second.
So we keep nodes small. At 8 TB a rebuild takes 15 hours, and a second failure in that window can lose a quorum.
You can change the numbers in the calculator. At 10 KB values, disk space takes over: 390 nodes for disk against 70 for traffic. At ten times the reads, the cluster needs 700 nodes, where splitting into several clusters starts to pay.
- Writes, peak
- 300,000/s
- Every write lands on N replicas: size the disks for replica writes, not client writes.
- Replica operations, peak (R = W = quorum)
- 2,100,000/s
- A quorum read touches a majority of replicas; a write touches all N.
- Nodes needed for traffic (60% busy)
- 70
- Raw storage: replicas and LSM overhead
- 86 TB
- 100 bytes per key for the key, timestamp and index; 1.3× for SSTables not yet compacted.
- Nodes needed for disk
- 43
- Cluster size
- 70
- At least three racks of N nodes, so placement can put each replica in its own rack.
- Disk nodes ÷ traffic nodes
- 0.61
- Below 1: traffic sets the node count, so buy cores and IOPS, not bigger disks.
- Data per node
- 1.2 TB
- A replacement node streams this much at about 150 MB/s: a couple of hours.
- Disk writes per node, with compaction (10×)
- 140 MB/s
- Leveled compaction rewrites each byte about 10 times. NVMe takes this with room to spare.
- Bloom filter RAM per node (10 bits a key)
- 1.1 GB
- About a 1% false positive rate, and it fits in RAM.
Build the design
Section titled “Build the design”We start with the simplest design that is correct. Then we let the numbers break it, one at a time. Each step fixes only what broke, adds only the boxes that fix needs, and says what the fix costs. Throughout, follow one scenario: a client writes a key, then reads it back.
Step 1. One node, a map and a log
Section titled “Step 1. One node, a map and a log”The smallest version is one server. It keeps every key in a hash map in memory. Each put is first appended to a log on disk and flushed, then applied to the map. After a crash, the server replays the log to rebuild the map.
Stage 1: One node, a map and a log
Stage 1 of 6
One node, a map and a log
Why this stage
Nothing yet.
This is the smallest design that stores and returns values.
What we add
One server with a hash map in RAM, and a write-ahead log it appends and fsyncs before acknowledging a put.
What it costs
Every key must fit in RAM, a restart replays the whole log, and one machine is one failure.
At 600k reads per second, Storage nodes is 300% busy. Stage 2 is the fix.
The numbers behind the busy bars
What one machine can do
- One in-memory server handles about 500,000 reads a second, but only about 100,000 durable writes, because it waits on fsync for its log.
- One LSM node (16 cores, NVMe) does about 40,000 reads a second that touch disk and 60,000 writes.
- Mixed, that is about 50,000 replica operations a second, including the work of coordinating requests. The coordinator is a role storage nodes play, not separate servers.
Why replicas add work
- From stage 4, each client write becomes 3 replica writes, and each quorum read becomes 2 replica reads.
- From stage 6, each data center applies every write, and is sized to take all reads if the other one fails.
- Instance counts keep each box near 60% busy.
It breaks on size. About 22 TB of keys and values don’t fit in the 512 GB of memory one server has, and replaying a log that big after a restart would take hours.
Go deeperThe client API1 min
The client API is small. What matters is what travels with each call: how many copies must answer, and a context that carries the version the client last read:
GET /v1/{table}/{key}?consistency=QUORUM → 200 { "values": ["..."], "context": "a8F3k..." }PUT /v1/{table}/{key}?consistency=QUORUM X-Context: a8F3k... <value bytes> → 204Two error codes look alike but mean opposite things. 503 Unavailable means we knew in advance
that too few copies were alive, so nothing was written. 504 Timeout means too few copies answered
in time. The write may be on none, some or all of them. It is unknown, not failed.
Step 2. An LSM tree on disk
Section titled “Step 2. An LSM tree on disk”Even split across many machines, each node will hold about 1.2 TB, far more than its memory. Writes are a third of the traffic, so they must be cheap. An LSM tree (log-structured merge tree) is built for exactly this:
- Commit log. Every write is first appended to a log on disk and flushed. If the node crashes, the log is replayed. One flush every 2 ms covers hundreds of writes.
- Memtable. A sorted table in memory. When it reaches 256 MB, it is written to disk in one go, as a sorted file.
- SSTables. These sorted files are never changed once written. A read checks the memtable, then the files from newest to oldest, and stops at the first match, which is the newest version.
- Bloom filters. A small filter per file, kept in memory, that answers “this key is definitely not here”. It skips 99% of the files that don’t have the key, so most reads open one file.
- Compaction. A background job merges files, keeps the newest version of each key, and frees the rest.
// One storage node's engine. Writes go to the commit log, then the memtable; a full memtable// becomes an immutable sorted file (an SSTable). Files, block indexes and checksums are left out.class LsmStore { interface CommitLog { void appendAndSync(String key, String value); // returns once fsynced (group commit) }
// Older SSTables are immutable, so a delete cannot remove the key from them. It writes a // marker that hides older versions until compaction drops both. static final String TOMBSTONE = "\u0000tombstone";
record SSTable(TreeMap<String, String> rows, Bloom bloom) {}
private final CommitLog log; private final int flushAt; // entries here; real engines flush at about 256 MB private TreeMap<String, String> memtable = new TreeMap<>(); // sorted, so a flush is one sequential write private final Deque<SSTable> sstables = new ArrayDeque<>(); // newest first
LsmStore(CommitLog log, int flushAt) { this.log = log; this.flushAt = flushAt; }
void put(String key, String value) { log.appendAndSync(key, value); // durable before acked: a crash replays the log into a new memtable memtable.put(key, value); // no read before the write, which is why LSM writes are cheap if (memtable.size() >= flushAt) { sstables.addFirst(seal(memtable)); // written once, never modified memtable = new TreeMap<>(); // the log up to here can now be recycled } }
void delete(String key) { put(key, TOMBSTONE); }
Optional<String> get(String key) { String v = memtable.get(key); // newest data first if (v == null) { for (SSTable t : sstables) { // then newest file to oldest: the first hit is the newest version if (!t.bloom().mightContain(key)) continue; // rules out ~99% of files lacking the key, no disk read v = t.rows().get(key); // real engine: sparse index, then one 4 KB block from disk if (v != null) break; } } return v == null || v.equals(TOMBSTONE) ? Optional.empty() : Optional.of(v); }
// Compaction merges files so reads check fewer of them and overwritten versions free their // space. A tombstone may be dropped only once every replica has surely seen it (the grace // period, 10 days by default). Dropped sooner, a replica that missed the delete would hand // the old value back during repair, and the deleted key would return. void compactAll(boolean gracePeriodOver) { TreeMap<String, String> merged = new TreeMap<>(); for (Iterator<SSTable> it = sstables.descendingIterator(); it.hasNext(); ) { merged.putAll(it.next().rows()); // oldest first, so newer versions overwrite older ones } if (gracePeriodOver) merged.values().removeIf(TOMBSTONE::equals); sstables.clear(); sstables.addFirst(seal(merged)); }
private static SSTable seal(TreeMap<String, String> rows) { Bloom bloom = new Bloom(Math.max(64, rows.size() * 10)); // 10 bits a key: about 1% false positives rows.keySet().forEach(bloom::add); return new SSTable(rows, bloom); }}
// A Bloom filter answers "definitely not here" or "maybe here", in about 1.2 bytes a key.class Bloom { private static final int PROBES = 7; // the best count for 10 bits a key private final BitSet bits; private final int size;
Bloom(int size) { this.size = size; this.bits = new BitSet(size); }
void add(String key) { for (int i = 0; i < PROBES; i++) bits.set(index(key, i)); }
boolean mightContain(String key) { for (int i = 0; i < PROBES; i++) if (!bits.get(index(key, i))) return false; return true; // all bits set: maybe here, or a false positive }
// Double hashing: two base hashes make all seven probe positions. Production uses two // halves of a 128-bit Murmur3; hashCode is enough to show the idea. private int index(String key, int i) { int h1 = key.hashCode(); int h2 = Integer.reverse(h1) | 1; return Math.floorMod(h1 + i * h2, size); }}class CommitLog(Protocol): def append_and_sync(self, key: str, value: str) -> None: ... # returns once fsynced
# Older SSTables are immutable, so a delete cannot remove the key from them. It writes a marker# that hides older versions until compaction drops both.TOMBSTONE = "\0tombstone"
@dataclassclass SSTable: rows: dict[str, str] # sorted by key when written, so a flush is one sequential write bloom: "Bloom"
# One storage node's engine. Writes go to the commit log, then the memtable; a full# memtable becomes an immutable sorted file (an SSTable). Real files and indexes left out.class LsmStore: def __init__(self, log: CommitLog, flush_at: int): self._log = log self._flush_at = flush_at # entries here; real engines flush at about 256 MB self._memtable: dict[str, str] = {} self._sstables: list[SSTable] = [] # newest first
def put(self, key: str, value: str) -> None: self._log.append_and_sync(key, value) # durable before acked: a crash replays the log self._memtable[key] = value # no read before the write, which is why LSM writes are cheap if len(self._memtable) >= self._flush_at: self._sstables.insert(0, seal(self._memtable)) # written once, never modified self._memtable = {} # the log up to here can now be recycled
def delete(self, key: str) -> None: self.put(key, TOMBSTONE)
def get(self, key: str) -> str | None: v = self._memtable.get(key) # newest data first if v is None: for t in self._sstables: # then newest file to oldest: the first hit is the newest if not t.bloom.might_contain(key): continue # rules out ~99% of files lacking the key, with no disk read v = t.rows.get(key) # real engine: sparse index, then one 4 KB block if v is not None: break return None if v is None or v == TOMBSTONE else v
# Merge files so reads check fewer of them and overwritten versions free their space. # A tombstone may be dropped only once every replica has surely seen it (the grace # period, 10 days by default). Dropped sooner, a replica that missed the delete would # hand the old value back during repair, and the deleted key would return. def compact_all(self, grace_period_over: bool) -> None: merged: dict[str, str] = {} for t in reversed(self._sstables): # oldest first, so newer versions overwrite older merged.update(t.rows) if grace_period_over: merged = {k: v for k, v in merged.items() if v != TOMBSTONE} self._sstables = [seal(merged)]
def seal(rows: dict[str, str]) -> SSTable: bloom = Bloom(max(64, len(rows) * 10)) # 10 bits a key: about 1% false positives for k in rows: bloom.add(k) return SSTable(dict(sorted(rows.items())), bloom)
# Answers "definitely not here" or "maybe here", in about 1.2 bytes a key.class Bloom: PROBES = 7 # the best count for 10 bits a key
def __init__(self, size: int): self._size = size self._bits = bytearray((size + 7) // 8)
def add(self, key: str) -> None: for i in self._indexes(key): self._bits[i // 8] |= 1 << (i % 8)
def might_contain(self, key: str) -> bool: # all bits set: maybe here, or a false positive return all(self._bits[i // 8] & (1 << (i % 8)) for i in self._indexes(key))
def _indexes(self, key: str): # Double hashing: two base hashes make all seven probe positions. Production uses # Murmur3; hash() is enough here because the filter never leaves this process. h1 = hash(key) h2 = hash((key, 1)) | 1 return ((h1 + i * h2) % self._size for i in range(self.PROBES))// One storage node's engine. Writes go to the commit log, then the memtable; a full memtable// becomes an immutable sorted file (an SSTable). Files, block indexes and checksums are left out.interface ICommitLog{ void AppendAndSync(string key, string value); // returns once fsynced (group commit)}
record SSTable(SortedDictionary<string, string> Rows, Bloom Bloom);
class LsmStore(ICommitLog log, int flushAt) // flushAt in entries; real engines flush at ~256 MB{ // Older SSTables are immutable, so a delete cannot remove the key from them. It writes a // marker that hides older versions until compaction drops both. const string Tombstone = "\0tombstone";
private SortedDictionary<string, string> _memtable = new(StringComparer.Ordinal); // sorted: a flush is one sequential write private readonly List<SSTable> _sstables = new(); // newest first
public void Put(string key, string value) { log.AppendAndSync(key, value); // durable before acked: a crash replays the log into a new memtable _memtable[key] = value; // no read before the write, which is why LSM writes are cheap if (_memtable.Count >= flushAt) { _sstables.Insert(0, Seal(_memtable)); // written once, never modified _memtable = new(StringComparer.Ordinal); // the log up to here can now be recycled } }
public void Delete(string key) => Put(key, Tombstone);
public string? Get(string key) { if (!_memtable.TryGetValue(key, out var v)) // newest data first { foreach (var t in _sstables) // then newest file to oldest: the first hit is the newest version { if (!t.Bloom.MightContain(key)) continue; // rules out ~99% of files lacking the key, no disk read if (t.Rows.TryGetValue(key, out v)) break; // real engine: sparse index, then one 4 KB block } } return v is null || v == Tombstone ? null : v; }
// Compaction merges files so reads check fewer of them and overwritten versions free their // space. A tombstone may be dropped only once every replica has surely seen it (the grace // period, 10 days by default). Dropped sooner, a replica that missed the delete would hand // the old value back during repair, and the deleted key would return. public void CompactAll(bool gracePeriodOver) { var merged = new SortedDictionary<string, string>(StringComparer.Ordinal); for (int i = _sstables.Count - 1; i >= 0; i--) // oldest first, so newer versions overwrite older foreach (var (k, val) in _sstables[i].Rows) merged[k] = val; if (gracePeriodOver) foreach (var k in merged.Where(kv => kv.Value == Tombstone).Select(kv => kv.Key).ToList()) merged.Remove(k); _sstables.Clear(); _sstables.Add(Seal(merged)); }
private static SSTable Seal(SortedDictionary<string, string> rows) { var bloom = new Bloom(Math.Max(64, rows.Count * 10)); // 10 bits a key: about 1% false positives foreach (var k in rows.Keys) bloom.Add(k); return new SSTable(rows, bloom); }}
// A Bloom filter answers "definitely not here" or "maybe here", in about 1.2 bytes a key.class Bloom(int size){ const int Probes = 7; // the best count for 10 bits a key private readonly System.Collections.BitArray _bits = new(size);
public void Add(string key) { for (int i = 0; i < Probes; i++) _bits[Index(key, i)] = true; }
public bool MightContain(string key) { for (int i = 0; i < Probes; i++) if (!_bits[Index(key, i)]) return false; return true; // all bits set: maybe here, or a false positive }
// Double hashing: two base hashes make all seven probe positions. Production uses // Murmur3; GetHashCode is enough here because the filter never leaves this process. private int Index(string key, int i) { int h1 = key.GetHashCode(); int h2 = HashCode.Combine(key, 1) | 1; return (int)((uint)(h1 + i * h2) % (uint)size); }}// CommitLog is the node's write-ahead log. AppendAndSync returns once the entry is fsynced.type CommitLog interface { AppendAndSync(key, value string) error}
// Older SSTables are immutable, so a delete cannot remove the key from them. It writes a// marker that hides older versions until compaction drops both.const tombstone = "\x00tombstone"
// SSTable is an immutable run of rows. On disk it is sorted by key, so a flush is one// sequential write; a map stands in for the file here.type SSTable struct { rows map[string]string bloom *Bloom}
// LsmStore is one storage node's engine. Writes go to the commit log, then the memtable; a// full memtable becomes an SSTable. Files, block indexes and checksums are left out.type LsmStore struct { log CommitLog flushAt int // entries here; real engines flush at about 256 MB memtable map[string]string // a sorted skip list in real engines sstables []*SSTable // newest first}
func (s *LsmStore) Put(key, value string) error { // Durable before acked: a crash replays the log into a new memtable. if err := s.log.AppendAndSync(key, value); err != nil { return err } s.memtable[key] = value // no read before the write, which is why LSM writes are cheap if len(s.memtable) >= s.flushAt { s.sstables = append([]*SSTable{seal(s.memtable)}, s.sstables...) // written once, never modified s.memtable = map[string]string{} // the log up to here can be recycled } return nil}
func (s *LsmStore) Delete(key string) error { return s.Put(key, tombstone) }
func (s *LsmStore) Get(key string) (string, bool) { v, ok := s.memtable[key] // newest data first if !ok { for _, t := range s.sstables { // then newest file to oldest: the first hit is the newest version if !t.bloom.MightContain(key) { continue // rules out ~99% of files lacking the key, with no disk read } if v, ok = t.rows[key]; ok { // real engine: sparse index, then one 4 KB block break } } } if !ok || v == tombstone { return "", false } return v, true}
// CompactAll merges files so reads check fewer of them and overwritten versions free their// space. A tombstone may be dropped only once every replica has surely seen it (the grace// period, 10 days by default). Dropped sooner, a replica that missed the delete would hand// the old value back during repair, and the deleted key would return.func (s *LsmStore) CompactAll(gracePeriodOver bool) { merged := map[string]string{} for i := len(s.sstables) - 1; i >= 0; i-- { // oldest first, so newer versions overwrite older for k, v := range s.sstables[i].rows { merged[k] = v } } if gracePeriodOver { for k, v := range merged { if v == tombstone { delete(merged, k) } } } s.sstables = []*SSTable{seal(merged)}}
func seal(rows map[string]string) *SSTable { b := NewBloom(max(64, len(rows)*10)) // 10 bits a key: about 1% false positives for k := range rows { b.Add(k) } return &SSTable{rows: rows, bloom: b}}
// Bloom answers "definitely not here" or "maybe here", in about 1.2 bytes a key.type Bloom struct { bits []uint64 size uint64}
const probes = 7 // the best count for 10 bits a key
func NewBloom(size int) *Bloom { return &Bloom{bits: make([]uint64, (size+63)/64), size: uint64(size)}}
func (b *Bloom) Add(key string) { for _, i := range b.indexes(key) { b.bits[i/64] |= 1 << (i % 64) }}
func (b *Bloom) MightContain(key string) bool { for _, i := range b.indexes(key) { if b.bits[i/64]&(1<<(i%64)) == 0 { return false } } return true // all bits set: maybe here, or a false positive}
// indexes uses double hashing: two halves of one 64-bit hash make all seven probe// positions. Production uses Murmur3; FNV is enough to show the idea.func (b *Bloom) indexes(key string) [probes]uint64 { h := fnv.New64a() h.Write([]byte(key)) sum := h.Sum64() h1, h2 := sum&0xffffffff, (sum>>32)|1 var out [probes]uint64 for i := range out { out[i] = (h1 + uint64(i)*h2) % b.size } return out}interface CommitLog { appendAndSync(key: string, value: string): Promise<void>; // resolves once fsynced}
// Older SSTables are immutable, so a delete cannot remove the key from them. It writes a// marker that hides older versions until compaction drops both.const TOMBSTONE = '\0tombstone';
// An immutable run of rows. On disk it is sorted by key, so a flush is one sequential write;// a Map stands in for the file here.interface SSTable { rows: Map<string, string>; bloom: Bloom;}
// One storage node's engine. Writes go to the commit log, then the memtable; a full memtable// becomes an SSTable. Files, block indexes and checksums are left out.class LsmStore { private memtable = new Map<string, string>(); // a sorted skip list in real engines private sstables: SSTable[] = []; // newest first
constructor( private readonly log: CommitLog, private readonly flushAt: number, // entries here; real engines flush at about 256 MB ) {}
async put(key: string, value: string): Promise<void> { await this.log.appendAndSync(key, value); // durable before acked: a crash replays the log this.memtable.set(key, value); // no read before the write, which is why LSM writes are cheap if (this.memtable.size >= this.flushAt) { this.sstables.unshift(seal(this.memtable)); // written once, never modified this.memtable = new Map(); // the log up to here can now be recycled } }
delete(key: string): Promise<void> { return this.put(key, TOMBSTONE); }
get(key: string): string | undefined { let v = this.memtable.get(key); // newest data first if (v === undefined) { for (const t of this.sstables) { // newest file to oldest: the first hit is the newest version if (!t.bloom.mightContain(key)) continue; // rules out ~99% of files lacking the key v = t.rows.get(key); // real engine: sparse index, then one 4 KB block if (v !== undefined) break; } } return v === TOMBSTONE ? undefined : v; }
// Compaction merges files so reads check fewer of them and overwritten versions free their // space. A tombstone may be dropped only once every replica has surely seen it (the grace // period, 10 days by default). Dropped sooner, a replica that missed the delete would hand // the old value back during repair, and the deleted key would return. compactAll(gracePeriodOver: boolean): void { const merged = new Map<string, string>(); for (const t of [...this.sstables].reverse()) { for (const [k, v] of t.rows) merged.set(k, v); // oldest first, newer versions overwrite } if (gracePeriodOver) for (const [k, v] of merged) if (v === TOMBSTONE) merged.delete(k); this.sstables = [seal(merged)]; }}
function seal(rows: Map<string, string>): SSTable { const bloom = new Bloom(Math.max(64, rows.size * 10)); // 10 bits a key: about 1% false positives for (const k of rows.keys()) bloom.add(k); return { rows, bloom };}
// A Bloom filter answers "definitely not here" or "maybe here", in about 1.2 bytes a key.class Bloom { private static readonly PROBES = 7; // the best count for 10 bits a key private readonly bits: Uint8Array;
constructor(private readonly size: number) { this.bits = new Uint8Array(Math.ceil(size / 8)); }
add(key: string): void { for (const i of this.indexes(key)) this.bits[i >> 3] |= 1 << (i & 7); }
mightContain(key: string): boolean { // all bits set: maybe here, or a false positive return this.indexes(key).every((i) => (this.bits[i >> 3] & (1 << (i & 7))) !== 0); }
// Double hashing: two base hashes make all seven probe positions. Production uses // Murmur3; two seeded FNV-1a hashes are enough to show the idea. private indexes(key: string): number[] { const h1 = fnv1a(key, 0x811c9dc5); const h2 = fnv1a(key, 0x01000193) | 1; return Array.from({ length: Bloom.PROBES }, (_, i) => (h1 + i * h2) % this.size); }}
function fnv1a(s: string, seed: number): number { let h = seed; for (let i = 0; i < s.length; i++) h = Math.imul(h ^ s.charCodeAt(i), 0x01000193); return h >>> 0; // unsigned 32 bits}Deletes are subtle. Files are never changed, so a delete writes a marker called a tombstone that hides older versions. Compaction may remove the tombstone only after a grace period (10 days in Cassandra). By then, every copy must have received it. Remove it sooner, and a copy that missed the delete can hand the old value back, and the key comes back from the dead.
Stage 2: An LSM tree on disk
Stage 2 of 6
An LSM tree on disk
Why this stage
About 22 TB of keys and values do not fit in the 512 GB of RAM one server has, and replaying a log that size on restart would take hours.
Data size, not load, forces this stage.
What we add
A sorted memtable that flushes to immutable SSTables on disk, Bloom filters per SSTable, and background compaction.
What it costs
Reads may check several files, compaction rewrites data many times, and deletes become tombstones.
At 600k reads per second, Storage nodes is 1500% busy. Stage 3 is the fix.
The numbers behind the busy bars
What one machine can do
- One in-memory server handles about 500,000 reads a second, but only about 100,000 durable writes, because it waits on fsync for its log.
- One LSM node (16 cores, NVMe) does about 40,000 reads a second that touch disk and 60,000 writes.
- Mixed, that is about 50,000 replica operations a second, including the work of coordinating requests. The coordinator is a role storage nodes play, not separate servers.
Why replicas add work
- From stage 4, each client write becomes 3 replica writes, and each quorum read becomes 2 replica reads.
- From stage 6, each data center applies every write, and is sized to take all reads if the other one fails.
- Instance counts keep each box near 60% busy.
Go deeperWhy an LSM tree, and which compaction2 min
| Engine | How it works | Good | Bad |
|---|---|---|---|
| Hash index + log (Bitcask) | Values in a log; a map of key → place in RAM | One disk read per get | Every key in RAM: about 86 GB per node here |
| B-tree (InnoDB, LMDB) | Sorted pages, changed in place | 2 to 3 page reads per get; predictable | Random writes; each small update rewrites a whole page |
| LSM tree (RocksDB, Cassandra) | Writes buffered in RAM, flushed as sorted files | Disk only appends; compresses well | Reads may check several files; compaction uses bandwidth |
There are two main ways to compact:
| Compaction | Rewrites each byte | Extra space | Files per read | Fits |
|---|---|---|---|---|
| Size-tiered | A few times | Up to 2× during merges | Several | Write-heavy, rarely updated |
| Leveled | 10 to 30 times | About 1.1× | At most one per level | Read-heavy, often overwritten |
We pick leveled, because reads outnumber writes 2 to 1, and keys like carts are overwritten constantly. The price is that compaction becomes the main user of disk bandwidth. If it falls behind, reads slow down gradually over hours, so we alert on it.
Why flush every 2 ms and not on every write? Flushing each write in turn manages only a few thousand writes a second. Cassandra’s default is weaker still: it flushes every 10 seconds and answers before that, betting that three copies in three racks won’t lose power together. It’s a fair bet, but make it on purpose.
Step 3. Spread keys with a hash ring
Section titled “Step 3. Spread keys with a hash ring”At the peak of 600,000 reads a second, one LSM node is more than ten times past what it can do, and
the data is more than ten times its disk. So the keys must be split across about 70 machines. The
simplest idea is hash(key) mod 70. But add one node, and the remainder changes for almost
every key. Nearly all the data would move.
Consistent hashing fixes this. Picture a ring of numbers. Each node owns some positions on it, called tokens. A key’s hash lands somewhere on the ring, and the first token clockwise owns it. Adding a node only takes over the keys just before its tokens, about 1/70 of the data.
With one token per node, the slices are uneven, and a new node takes all its load from one neighbour. So each machine takes 16 tokens, called virtual nodes. It owns 16 small slices, and a new machine takes a little from many others.
A key’s 3 copies are found by walking clockwise and taking the first 3 different machines, each in a different rack. Different machines, because two tokens on one machine fail together. Different racks, because one broken network switch takes down a whole rack.
// The token ring: which N nodes own a key. Every node and every client library holds a copy,// built from the membership that gossip spreads, so routing a request needs no lookup service.class Ring { record Node(String id, String rack) {}
private final TreeMap<Long, Node> tokens = new TreeMap<>(); // token -> owner, in ring order private final int vnodes; // tokens per physical node: 16 here, Cassandra 4's default
Ring(int vnodes) { this.vnodes = vnodes; }
// Many small ranges per node instead of one big one. A joining node takes a slice from // every node rather than half of one neighbor, and a dead node's load spreads the same way. void add(Node node) { for (int i = 0; i < vnodes; i++) tokens.put(hash(node.id() + "#" + i), node); }
void remove(Node node) { tokens.values().removeIf(node::equals); }
// The preference list: walk clockwise from the key's token and keep the first n distinct // nodes, each in a rack not used yet, so losing a whole rack costs a key one replica at most. List<Node> replicasFor(String key, int n) { List<Node> out = new ArrayList<>(); Set<String> racks = new HashSet<>(); // First pass insists on distinct racks; if there are fewer racks than n, the second // pass settles for distinct nodes. Two vnodes of one machine never count twice. for (boolean distinctRacks : new boolean[] {true, false}) { for (Node node : clockwiseFrom(hash(key))) { if (out.size() == n) return out; if (out.contains(node) || (distinctRacks && racks.contains(node.rack()))) continue; out.add(node); racks.add(node.rack()); } } return out; // fewer than n only when the cluster has fewer than n nodes }
// Tokens at or after t, then wrap to the start: that wrap is what makes it a ring. // Copying keeps the idea visible; production walks the sorted token array in place. private List<Node> clockwiseFrom(long t) { List<Node> walk = new ArrayList<>(tokens.tailMap(t).values()); walk.addAll(tokens.headMap(t).values()); return walk; }
// MD5 spreads tokens evenly and is the same in every language (Cassandra uses Murmur3). // Never String.hashCode: every client must compute exactly the token the servers compute. static long hash(String s) { try { byte[] d = MessageDigest.getInstance("MD5").digest(s.getBytes(StandardCharsets.UTF_8)); return ByteBuffer.wrap(d).getLong(); // first 8 bytes, big-endian, signed } catch (NoSuchAlgorithmException e) { throw new IllegalStateException(e); } }}# The token ring: which N nodes own a key. Every node and every client library holds a# copy, built from the membership that gossip spreads, so routing needs no lookup service.class Ring: def __init__(self, vnodes: int = 16): # tokens per physical node, Cassandra 4's default self._vnodes = vnodes self._tokens: list[int] = [] # sorted: the ring order self._owner: dict[int, Node] = {} # token -> physical node
def add(self, node: "Node") -> None: # Many small ranges per node instead of one big one. A joining node takes a slice from # every node rather than half of one neighbor, and a dead node's load spreads the same way. for i in range(self._vnodes): t = token(f"{node.id}#{i}") bisect.insort(self._tokens, t) self._owner[t] = node
def remove(self, node: "Node") -> None: self._tokens = [t for t in self._tokens if self._owner[t] != node] self._owner = {t: self._owner[t] for t in self._tokens}
# The preference list: walk clockwise from the key's token and keep the first n # distinct nodes, each in a rack not used yet, so losing a rack costs at most one replica. def replicas_for(self, key: str, n: int) -> list["Node"]: out: list[Node] = [] racks: set[str] = set() # First pass insists on distinct racks; if there are fewer racks than n, the second # pass settles for distinct nodes. Two vnodes of one machine never count twice. for distinct_racks in (True, False): for node in self._clockwise_from(token(key)): if len(out) == n: return out if node in out or (distinct_racks and node.rack in racks): continue out.append(node) racks.add(node.rack) return out # fewer than n only when the cluster has fewer than n nodes
def _clockwise_from(self, t: int): # Tokens at or after t, then wrap to the start: that wrap is what makes it a ring. i = bisect.bisect_left(self._tokens, t) for tok in self._tokens[i:] + self._tokens[:i]: yield self._owner[tok]
@dataclass(frozen=True)class Node: id: str rack: str
def token(s: str) -> int: # MD5 spreads tokens evenly and is the same in every language (Cassandra uses Murmur3). # Never hash(): Python salts it per process, and every client must compute the same token. d = hashlib.md5(s.encode()).digest() return int.from_bytes(d[:8], "big", signed=True)// The token ring: which N nodes own a key. Every node and every client library holds a copy,// built from the membership that gossip spreads, so routing a request needs no lookup service.class Ring(int vnodes = 16) // tokens per physical node: Cassandra 4's default{ public record Node(string Id, string Rack);
private readonly SortedList<long, Node> _tokens = new(); // token -> owner, in ring order
// Many small ranges per node instead of one big one. A joining node takes a slice from // every node rather than half of one neighbor, and a dead node's load spreads the same way. public void Add(Node node) { for (int i = 0; i < vnodes; i++) _tokens[Hash($"{node.Id}#{i}")] = node; }
public void Remove(Node node) { foreach (var t in _tokens.Where(kv => kv.Value == node).Select(kv => kv.Key).ToList()) _tokens.Remove(t); }
// The preference list: walk clockwise from the key's token and keep the first n distinct // nodes, each in a rack not used yet, so losing a whole rack costs a key one replica at most. public List<Node> ReplicasFor(string key, int n) { var output = new List<Node>(); var racks = new HashSet<string>(); // First pass insists on distinct racks; if there are fewer racks than n, the second // pass settles for distinct nodes. Two vnodes of one machine never count twice. foreach (bool distinctRacks in new[] { true, false }) { foreach (var node in ClockwiseFrom(Hash(key))) { if (output.Count == n) return output; if (output.Contains(node) || (distinctRacks && racks.Contains(node.Rack))) continue; output.Add(node); racks.Add(node.Rack); } } return output; // fewer than n only when the cluster has fewer than n nodes }
// Tokens at or after t, then wrap to the start: that wrap is what makes it a ring. // A linear scan keeps the idea visible; production binary-searches the sorted tokens. private IEnumerable<Node> ClockwiseFrom(long t) { var after = _tokens.Where(kv => kv.Key >= t).Select(kv => kv.Value); var before = _tokens.Where(kv => kv.Key < t).Select(kv => kv.Value); return after.Concat(before); }
// MD5 spreads tokens evenly and is the same in every language (Cassandra uses Murmur3). // Never string.GetHashCode: .NET randomizes it per process, and every client must // compute exactly the token the servers compute. public static long Hash(string s) { byte[] d = System.Security.Cryptography.MD5.HashData(Encoding.UTF8.GetBytes(s)); return System.Buffers.Binary.BinaryPrimitives.ReadInt64BigEndian(d); // first 8 bytes }}// Node is one physical machine. Rack is its failure domain.type Node struct { ID string Rack string}
// Ring maps a key to the N nodes that own it. Every node and every client library holds a// copy, built from the membership that gossip spreads, so routing needs no lookup service.type Ring struct { vnodes int // tokens per physical node: 16, Cassandra 4's default tokens []int64 // sorted: the ring order owner map[int64]Node // token -> physical node}
func NewRing(vnodes int) *Ring { return &Ring{vnodes: vnodes, owner: map[int64]Node{}}}
// Add gives the node many small ranges instead of one big one. A joining node takes a slice// from every node rather than half of one neighbor, and a dead node's load spreads the same way.func (r *Ring) Add(n Node) { for i := 0; i < r.vnodes; i++ { t := token(fmt.Sprintf("%s#%d", n.ID, i)) r.owner[t] = n r.tokens = append(r.tokens, t) } sort.Slice(r.tokens, func(i, j int) bool { return r.tokens[i] < r.tokens[j] })}
func (r *Ring) Remove(n Node) { kept := r.tokens[:0] for _, t := range r.tokens { if r.owner[t] == n { delete(r.owner, t) } else { kept = append(kept, t) } } r.tokens = kept}
// ReplicasFor is the preference list: walk clockwise from the key's token and keep the first// n distinct nodes, each in a rack not used yet, so losing a rack costs at most one replica.func (r *Ring) ReplicasFor(key string, n int) []Node { var out []Node racks := map[string]bool{} taken := map[Node]bool{} start := sort.Search(len(r.tokens), func(i int) bool { return r.tokens[i] >= token(key) }) // First pass insists on distinct racks; if there are fewer racks than n, the second pass // settles for distinct nodes. Two vnodes of one machine never count twice. for _, distinctRacks := range []bool{true, false} { for i := 0; i < len(r.tokens); i++ { if len(out) == n { return out } // The modulo is the wrap from the last token back to the first: the ring. node := r.owner[r.tokens[(start+i)%len(r.tokens)]] if taken[node] || (distinctRacks && racks[node.Rack]) { continue } out = append(out, node) taken[node], racks[node.Rack] = true, true } } return out // fewer than n only when the cluster has fewer than n nodes}
// token uses MD5 because it spreads evenly and is the same in every language (Cassandra// uses Murmur3). Every client must compute exactly the token the servers compute.func token(s string) int64 { d := md5.Sum([]byte(s)) return int64(binary.BigEndian.Uint64(d[:8])) // first 8 bytes, big-endian, signed}interface RingNode { id: string; rack: string; // its failure domain}
// The token ring: which N nodes own a key. Every node and every client library holds a copy,// built from the membership that gossip spreads, so routing a request needs no lookup service.class Ring { private tokens: { token: bigint; node: RingNode }[] = []; // sorted: the ring order
constructor(private readonly vnodes = 16) {} // tokens per physical node, Cassandra 4's default
// Many small ranges per node instead of one big one. A joining node takes a slice from // every node rather than half of one neighbor, and a dead node's load spreads the same way. add(node: RingNode): void { for (let i = 0; i < this.vnodes; i++) this.tokens.push({ token: token(`${node.id}#${i}`), node }); this.tokens.sort((a, b) => (a.token < b.token ? -1 : a.token > b.token ? 1 : 0)); }
remove(node: RingNode): void { this.tokens = this.tokens.filter((t) => t.node.id !== node.id); }
// The preference list: walk clockwise from the key's token and keep the first n distinct // nodes, each in a rack not used yet, so losing a whole rack costs a key one replica at most. replicasFor(key: string, n: number): RingNode[] { const out: RingNode[] = []; const racks = new Set<string>(); const t = token(key); // Tokens at or after t, then wrap to the start: that wrap is what makes it a ring. // A linear search keeps the idea visible; production binary-searches. let start = this.tokens.findIndex((x) => x.token >= t); if (start < 0) start = 0; // First pass insists on distinct racks; if there are fewer racks than n, the second pass // settles for distinct nodes. Two vnodes of one machine never count twice. for (const distinctRacks of [true, false]) { for (let i = 0; i < this.tokens.length; i++) { if (out.length === n) return out; const node = this.tokens[(start + i) % this.tokens.length].node; if (out.some((o) => o.id === node.id) || (distinctRacks && racks.has(node.rack))) continue; out.push(node); racks.add(node.rack); } } return out; // fewer than n only when the cluster has fewer than n nodes }}
// MD5 spreads tokens evenly and is the same in every language (Cassandra uses Murmur3).// Every client must compute exactly the token the servers compute.function token(s: string): bigint { return createHash('md5').update(s).digest().readBigInt64BE(0); // first 8 bytes, signed}Stage 3: Partition the keyspace
Stage 3 of 6
Partition the keyspace
Why this stage
At the 600,000 reads/s peak one LSM node is past 1,000% busy, and the data is more than 10 times its usable disk.
The keys have to be split across many machines.
What we add
A consistent hash ring with virtual nodes, a coordinator role (played by whichever node receives the request) that routes each key to its owners, and a gossip layer that tells every node who owns what.
What it costs
A request may take one more hop, and every node failure now makes a slice of keys unreachable.
At 600k reads per second, every part has headroom. Raise the traffic until something turns red.
The numbers behind the busy bars
What one machine can do
- One in-memory server handles about 500,000 reads a second, but only about 100,000 durable writes, because it waits on fsync for its log.
- One LSM node (16 cores, NVMe) does about 40,000 reads a second that touch disk and 60,000 writes.
- Mixed, that is about 50,000 replica operations a second, including the work of coordinating requests. The coordinator is a role storage nodes play, not separate servers.
Why replicas add work
- From stage 4, each client write becomes 3 replica writes, and each quorum read becomes 2 replica reads.
- From stage 6, each data center applies every write, and is sized to take all reads if the other one fails.
- Instance counts keep each box near 60% busy.
Go deeperHow clients find nodes, and how nodes notice a failure1 min
How does a client find the right nodes? The client library keeps a copy of the ring and sends each request straight to a copy of the key. That node acts as the coordinator: it sends the request to all 3 copies and collects the answers. Nodes learn about each other by gossip: every second, each node swaps news with a few random others. News reaches all 70 nodes in about 7 rounds.
A fixed timeout either reacts late or raises false alarms. So nodes use a phi accrual detector. Each node learns how often it normally hears from each peer. When the silence becomes very unlikely, it marks the peer as down, usually within 10 seconds.
A “down” mark only changes where requests go. It never changes who owns which keys. Ownership only changes when an operator adds or removes a node. Otherwise, a node in a long garbage collection pause could trigger a huge, needless data move.
Other ways to route requests exist. A separate proxy tier adds a hop to every request. Letting any node coordinate means the node rarely owns the key, which also adds a hop. The token-aware client adds none, but each language needs a library that tracks the ring.
Step 4. Copies with quorums
Section titled “Step 4. Copies with quorums”With 70 nodes, a slow or dead node is a weekly event. So each key has N = 3 copies.
There are two broad ways to keep copies. One gives each group of keys a leader that orders all writes, using a protocol like Raft. Reads are always up to date, but when a leader fails, its keys can’t be written for a few seconds. The other has no leader: any copy takes writes. We pick no leader, because a write must succeed whenever 2 copies are reachable, with no pause.
So the node that receives a write, called the coordinator, sends it to all 3 copies, and answers after W = 2 of them confirm. A read asks R = 2 copies. Because 2 + 2 is more than 3, every pair of copies a read can ask shares at least one copy with every pair a write reached. Follow one write and one read through the design:
Stage 4: Replicate with quorums
Stage 4 of 6
Replicate with quorums
Why this stage
With 70 nodes, a slow, restarting or dead node is a weekly event.
Without copies, each failure loses data and makes its keys unavailable, against a 99.99% target.
A failure domain, not load, forces this stage.
What we add
Each key lives on N = 3 nodes in 3 racks.
The coordinator sends every write to all 3 and waits for W = 2; reads wait for R = 2.
What it costs
Three times the writes and twice the reads at the disks, and replicas that can disagree.
At 600k reads per second, every part has headroom. Raise the traffic until something turns red.
The numbers behind the busy bars
What one machine can do
- One in-memory server handles about 500,000 reads a second, but only about 100,000 durable writes, because it waits on fsync for its log.
- One LSM node (16 cores, NVMe) does about 40,000 reads a second that touch disk and 60,000 writes.
- Mixed, that is about 50,000 replica operations a second, including the work of coordinating requests. The coordinator is a role storage nodes play, not separate servers.
Why replicas add work
- From stage 4, each client write becomes 3 replica writes, and each quorum read becomes 2 replica reads.
- From stage 6, each data center applies every write, and is sized to take all reads if the other one fails.
- Instance counts keep each box near 60% busy.
Trace a request: Quorum write (W = 2)
Quorum write (W = 2): step 1 of 5
App clients to Coordinator
The client library hashes the key with Murmur3 to a 64-bit token, looks up its 3 replicas in a cached copy of the ring, and sends the PUT straight to one of them. That replica acts as coordinator, so there is no extra hop.
- App clients → Coordinator0.5 ms
- Coordinator → Storage nodes0.5 ms
- Storage nodes → Commit log1.5 ms
- Storage nodes → Memtable0.1 ms
- Response
Waiting for 2 of 3 also keeps responses fast. A copy that’s briefly stuck, say in a garbage collection pause, is simply not one of the first two to answer.
This is enough for a mid-level answer. An LSM tree per node, consistent hashing with virtual nodes, and 3 copies with quorum reads and writes, each with a reason. Step 5 and the stress tests are what make it a senior answer.
Go deeperOther quorum settings1 min
Callers can choose other settings per request. W = 1 and R = 1 survive two copies down, for logs and metrics where speed matters more than freshness. W = 3 and R = 1 make reads cheap but let writes block.
Step 5. Repairing drift
Section titled “Step 5. Repairing drift”The last thing to break is the copies themselves. Say a node was down for six hours. How do we find what it missed, without comparing 1.2 TB?
At 70 nodes, something is always failing. Every failure leaves copies that disagree. So we need tools that bring them back together. There are three, each for a different case:
- Hinted handoff. When a copy is down, the coordinator writes to the 2 live ones, and saves the missed write as a hint. When the node comes back, the hint is replayed. This heals outages shorter than 3 hours. Hints don’t count towards W, which is what keeps our quorums strict.
- Read repair. When a read finds two copies that disagree, the coordinator writes the newer version to the older copy, then answers. This heals keys that get read.
- Anti-entropy. A scheduled job compares whole copies, for everything else, including missed deletes.
Stage 5: Repair what failures leave behind
Stage 5 of 6
Repair what failures leave behind
Why this stage
At 70 nodes something is always down, slow or restarting.
Every write it misses makes one replica stale, and stale replicas resurrect deleted data if nothing repairs them.
What we add
Hinted handoff for short outages, read repair on every digest mismatch, and a scheduled anti-entropy job that compares Merkle trees.
What it costs
Hints use disk on the coordinators, and repair streams compete with live traffic for disk and network.
At 600k reads per second, every part has headroom. Raise the traffic until something turns red.
The numbers behind the busy bars
What one machine can do
- One in-memory server handles about 500,000 reads a second, but only about 100,000 durable writes, because it waits on fsync for its log.
- One LSM node (16 cores, NVMe) does about 40,000 reads a second that touch disk and 60,000 writes.
- Mixed, that is about 50,000 replica operations a second, including the work of coordinating requests. The coordinator is a role storage nodes play, not separate servers.
Why replicas add work
- From stage 4, each client write becomes 3 replica writes, and each quorum read becomes 2 replica reads.
- From stage 6, each data center applies every write, and is sized to take all reads if the other one fails.
- Instance counts keep each box near 60% busy.
- Color key
- Database
How do we compare 1.2 TB of data cheaply? With a Merkle tree. Each leaf is a hash of the keys in a small slice of the ring. Each parent is a hash of its children. Two copies compare their top hashes. If they match, the data is the same and nothing more is sent. If not, they look only at the children that differ, and so on down. A few differing keys are found in about 15 steps.
Repair has a deadline. Every slice must be repaired more often than the 10-day tombstone grace period. If not, a copy that missed a delete keeps the old value. The others drop their tombstones during compaction. Then the next repair copies the deleted value back everywhere.
Go deeperFollow a hinted write, and what repair costs1 min
Trace a request: Write with a replica down
Write with a replica down: step 1 of 5
App clients to Coordinator
A PUT for a key whose replicas are nodes A, B and C. Node C’s disk controller hung 40 seconds ago.
- App clients → Coordinator0.5 ms
- Coordinator → Gossip layer0 ms
- Coordinator → Storage nodes2 ms
- Coordinator → Hint storeasync
- Response
Read repair waits for its fix before answering. That way, once a quorum read returns a value, no later quorum read can return an older one.
Merkle repair has two costs. Building the trees means reading every key in the slice, which costs disk. And a leaf still covers hundreds of keys, so one differing key sends hundreds over the network. So we run incremental repair, which only reads data written since the last repair. It runs all the time, slowly, so each slice is repaired about every 5 days.
Repair isn’t only for outages. Copies drift without any node going down: a timed-out write, a dropped hint, a write shed under load. So repair is routine.
Stress it
Section titled “Stress it”The design is complete. Now we test what it really promises: what R + W > N does and doesn’t guarantee, two writes that clash, and adding nodes without trouble. No new boxes appear here.
What R + W > N does and does not promise
Section titled “What R + W > N does and does not promise”You chose R = 2 and W = 2 with N = 3. So reads always see the latest write, right?
Whether you know exactly which promises quorums make, and which they don’t, such as reading your own writes after a timeout.
What it promises. If a write reached 2 of the 3 copies, every read of 2 copies includes at least one of them. The coordinator returns the newest version it sees. So a read that starts after a write succeeded returns that write, or something newer.
What it does not promise:
| Case | What happens | Our answer |
|---|---|---|
| Sloppy quorum | Acks come from stand-in nodes outside the key’s 3 copies; a read of the 3 misses them | Strict quorums: stand-ins never count |
| A write that timed out | It may sit on one copy, be missed by reads, then appear later after repair | Clients treat 504 as unknown |
| Latest timestamp wins, bad clocks | A later write from a clock 20 ms slow loses to an earlier one, silently | Keep both versions where that matters |
| Read, change, write | Two clients read 5 and both write 6. One increment is lost | Compare-and-set through Paxos (later) |
| A node joining | The copy set changes, so old and new quorums may not overlap | One extra ack while joining; one change at a time |
The sloppy quorum is the case people miss. The original Dynamo chose it on purpose, so a cart could always be written. When the key’s own copies were unreachable, a stand-in node took the write and counted towards W. The price is that 2 + 2 > 3 no longer means overlap:
- Color key
- Service
- Database
We take the other side. If 2 of a key’s 3 copies are unreachable, a QUORUM write fails with 503,
rather than landing somewhere else. A table that prefers Dynamo’s choice can write at W = 1, which
is honest about what it promises.
Go deeperCAP and PACELC, said precisely1 min
CAP’s “consistency” means linearizability. Its “availability” means every request to a working node gets an answer. During a network split, this design at ONE answers wherever one copy is reachable, and is not consistent. At QUORUM it is neither: the smaller side refuses, and the larger side still isn’t linearizable.
PACELC adds what happens when there is no split. This design is PA/EL, like Dynamo and Cassandra. During a split it keeps serving where it can (A). When healthy, it chooses low latency (L) over linearizability. Callers buy more consistency per request, with latency. Only compare-and-set through Paxos is consistent in both cases.
So “Cassandra is AP” isn’t quite right. Labels describe settings, not products. Say which operation, at which setting, and which odd result a user can see.
Two writes that clash
Section titled “Two writes that clash”Two users add items to the same cart at the same moment, in different datacenters. What’s in the cart afterwards?
With no leader, two clients can write the same key at the same time, through different copies. Both writes succeed. Later the copies meet and must decide what the key holds. There are a few ways:
- Last write wins. Each write carries a timestamp, and the highest wins. It’s simple, but one write is silently lost. And if a server’s clock is 20 ms slow, its later write can lose to an earlier one.
- Keep both. The store notices the two writes didn’t know about each other, and keeps both versions, called siblings. The next reader gets both and merges them.
- Merge automatically. Special data types, called CRDTs, merge themselves correctly, like a set where merging means union. They only fit data shaped that way.
- Never clash. Order every write to a key through Paxos. Correct, but slow.
We choose per table. Last write wins by default, for data where losing a clashing write is fine: sessions, profile fields, flags. Siblings where every write matters and the application can merge, like the cart.
How does the store know two writes didn’t see each other? With a version vector: a small list of counters, one per node that coordinated a write to the key. If every counter in version X is at least as high as in Y, then X came after Y and replaces it. If neither is ahead, the writes happened at the same time, and both are kept.
- Color key
- Service
// A version vector: one counter per node that coordinated a write to this key. It tells// "this version saw that one" apart from "these two were written without seeing each other",// which a timestamp cannot do.record VClock(Map<String, Long> counters) { static VClock empty() { return new VClock(Map.of()); }
// The coordinator bumps its own entry on every write it coordinates. VClock tick(String node) { Map<String, Long> next = new HashMap<>(counters); next.merge(node, 1L, Long::sum); return new VClock(next); }
// This version has seen everything `other` has: every counter at least as high. boolean descends(VClock other) { return other.counters.entrySet().stream() .allMatch(e -> counters.getOrDefault(e.getKey(), 0L) >= e.getValue()); }
// The client's context after reading siblings: the entry-wise maximum of their clocks. // A write made with it descends from every sibling, which is how siblings collapse. VClock merge(VClock other) { Map<String, Long> next = new HashMap<>(counters); other.counters.forEach((k, v) -> next.merge(k, v, Math::max)); return new VClock(next); }}
record Versioned(String value, VClock clock) {}
class Reconcile { // Run by the coordinator on the versions its R replicas returned. A version that another // one descends from is obsolete and dropped. What remains are siblings, written // concurrently: the store returns all of them, because only the application knows how to // merge two shopping carts. Last-write-wins would silently keep one and lose the other. static List<Versioned> latest(List<Versioned> versions) { List<Versioned> out = new ArrayList<>(); for (Versioned v : versions) { boolean obsolete = versions.stream().anyMatch(o -> o.clock().descends(v.clock()) && !v.clock().descends(o.clock())); // Equal clocks are the same version read from two replicas: keep one copy. boolean duplicate = out.stream().anyMatch(o -> o.clock().descends(v.clock()) && v.clock().descends(o.clock())); if (!obsolete && !duplicate) out.add(v); } return out; }
// The client writes back with the merged context of everything it read, so the new // version supersedes every sibling. Clocks grow by one entry per coordinator; Dynamo // trimmed the oldest entries past 10, which can rarely turn an ancestor into a false sibling. static Versioned write(String value, List<Versioned> readContext, String coordinator) { VClock ctx = readContext.stream().map(Versioned::clock).reduce(VClock.empty(), VClock::merge); return new Versioned(value, ctx.tick(coordinator)); }}# A version vector: one counter per node that coordinated a write to this key. It tells# "this version saw that one" apart from "these two were written without seeing each# other", which a timestamp cannot do.@dataclass(frozen=True)class VClock: counters: dict[str, int] = field(default_factory=dict)
def tick(self, node: str) -> "VClock": # The coordinator bumps its own entry on every write it coordinates. return VClock({**self.counters, node: self.counters.get(node, 0) + 1})
def descends(self, other: "VClock") -> bool: # This version has seen everything `other` has: every counter at least as high. return all(self.counters.get(k, 0) >= v for k, v in other.counters.items())
def merge(self, other: "VClock") -> "VClock": # The client's context after reading siblings: the entry-wise maximum of their clocks. # A write made with it descends from every sibling, which is how siblings collapse. keys = self.counters.keys() | other.counters.keys() return VClock({k: max(self.counters.get(k, 0), other.counters.get(k, 0)) for k in keys})
@dataclass(frozen=True)class Versioned: value: str clock: VClock
# Run by the coordinator on the versions its R replicas returned. A version that another# one descends from is obsolete and dropped. What remains are siblings, written# concurrently: the store returns all of them, because only the application knows how to# merge two shopping carts. Last-write-wins would silently keep one and lose the other.def latest(versions: list[Versioned]) -> list[Versioned]: out: list[Versioned] = [] for v in versions: obsolete = any(o.clock.descends(v.clock) and not v.clock.descends(o.clock) for o in versions) # Equal clocks are the same version read from two replicas: keep one copy. duplicate = any(o.clock.descends(v.clock) and v.clock.descends(o.clock) for o in out) if not obsolete and not duplicate: out.append(v) return out
def write(value: str, read_context: list[Versioned], coordinator: str) -> Versioned: # The client writes back with the merged context of everything it read, so the new # version supersedes every sibling. Clocks grow by one entry per coordinator; Dynamo # trimmed the oldest entries past 10, which can rarely turn an ancestor into a false sibling. ctx = VClock() for v in read_context: ctx = ctx.merge(v.clock) return Versioned(value, ctx.tick(coordinator))// A version vector: one counter per node that coordinated a write to this key. It tells// "this version saw that one" apart from "these two were written without seeing each other",// which a timestamp cannot do.record VClock(ImmutableDictionary<string, long> Counters){ public static readonly VClock Empty = new(ImmutableDictionary<string, long>.Empty);
// The coordinator bumps its own entry on every write it coordinates. public VClock Tick(string node) => new(Counters.SetItem(node, At(Counters, node) + 1));
// This version has seen everything `other` has: every counter at least as high. public bool Descends(VClock other) => other.Counters.All(e => At(Counters, e.Key) >= e.Value);
// The client's context after reading siblings: the entry-wise maximum of their clocks. // A write made with it descends from every sibling, which is how siblings collapse. public VClock Merge(VClock other) => new(other.Counters.Aggregate(Counters, (acc, e) => acc.SetItem(e.Key, Math.Max(At(acc, e.Key), e.Value))));
// A node missing from the clock has coordinated no write to this key: counter 0. static long At(ImmutableDictionary<string, long> c, string node) => c.TryGetValue(node, out var n) ? n : 0;}
record Versioned(string Value, VClock Clock);
static class Reconcile{ // Run by the coordinator on the versions its R replicas returned. A version that another // one descends from is obsolete and dropped. What remains are siblings, written // concurrently: the store returns all of them, because only the application knows how to // merge two shopping carts. Last-write-wins would silently keep one and lose the other. public static List<Versioned> Latest(IReadOnlyList<Versioned> versions) { var output = new List<Versioned>(); foreach (var v in versions) { bool obsolete = versions.Any(o => o.Clock.Descends(v.Clock) && !v.Clock.Descends(o.Clock)); // Equal clocks are the same version read from two replicas: keep one copy. bool duplicate = output.Any(o => o.Clock.Descends(v.Clock) && v.Clock.Descends(o.Clock)); if (!obsolete && !duplicate) output.Add(v); } return output; }
// The client writes back with the merged context of everything it read, so the new // version supersedes every sibling. Clocks grow by one entry per coordinator; Dynamo // trimmed the oldest entries past 10, which can rarely turn an ancestor into a false sibling. public static Versioned Write(string value, IEnumerable<Versioned> readContext, string coordinator) { var ctx = readContext.Select(v => v.Clock).Aggregate(VClock.Empty, (a, b) => a.Merge(b)); return new Versioned(value, ctx.Tick(coordinator)); }}// VClock is a version vector: one counter per node that coordinated a write to this key. It// tells "this version saw that one" apart from "these two were written without seeing each// other", which a timestamp cannot do.type VClock map[string]uint64
// Tick returns a copy with the coordinator's entry bumped, done on every write it coordinates.func (c VClock) Tick(node string) VClock { next := VClock{} for k, v := range c { next[k] = v } next[node]++ return next}
// Descends reports whether c has seen everything other has: every counter at least as high.func (c VClock) Descends(other VClock) bool { for k, v := range other { if c[k] < v { return false } } return true}
// Merge is the client's context after reading siblings: the entry-wise maximum of their// clocks. A write made with it descends from every sibling, which is how siblings collapse.func (c VClock) Merge(other VClock) VClock { next := VClock{} for k, v := range c { next[k] = v } for k, v := range other { next[k] = max(next[k], v) } return next}
type Versioned struct { Value string Clock VClock}
// Latest runs on the coordinator over the versions its R replicas returned. A version that// another one descends from is obsolete and dropped. What remains are siblings, written// concurrently: the store returns all of them, because only the application knows how to// merge two shopping carts. Last-write-wins would silently keep one and lose the other.func Latest(versions []Versioned) []Versioned { var out []Versioned for _, v := range versions { obsolete, duplicate := false, false for _, o := range versions { obsolete = obsolete || (o.Clock.Descends(v.Clock) && !v.Clock.Descends(o.Clock)) } // Equal clocks are the same version read from two replicas: keep one copy. for _, o := range out { duplicate = duplicate || (o.Clock.Descends(v.Clock) && v.Clock.Descends(o.Clock)) } if !obsolete && !duplicate { out = append(out, v) } } return out}
// Write builds the new version from the merged context of everything the client read, so it// supersedes every sibling. Clocks grow by one entry per coordinator; Dynamo trimmed the// oldest entries past 10, which can rarely turn an ancestor into a false sibling.func Write(value string, readContext []Versioned, coordinator string) Versioned { ctx := VClock{} for _, v := range readContext { ctx = ctx.Merge(v.Clock) } return Versioned{Value: value, Clock: ctx.Tick(coordinator)}}// A version vector: one counter per node that coordinated a write to this key. It tells// "this version saw that one" apart from "these two were written without seeing each other",// which a timestamp cannot do.class VClock { constructor(readonly counters: ReadonlyMap<string, number> = new Map()) {}
// The coordinator bumps its own entry on every write it coordinates. tick(node: string): VClock { return new VClock(new Map(this.counters).set(node, (this.counters.get(node) ?? 0) + 1)); }
// This version has seen everything `other` has: every counter at least as high. descends(other: VClock): boolean { return [...other.counters].every(([k, v]) => (this.counters.get(k) ?? 0) >= v); }
// The client's context after reading siblings: the entry-wise maximum of their clocks. // A write made with it descends from every sibling, which is how siblings collapse. merge(other: VClock): VClock { const next = new Map(this.counters); for (const [k, v] of other.counters) next.set(k, Math.max(next.get(k) ?? 0, v)); return new VClock(next); }}
interface Versioned { value: string; clock: VClock;}
// Run by the coordinator on the versions its R replicas returned. A version that another one// descends from is obsolete and dropped. What remains are siblings, written concurrently: the// store returns all of them, because only the application knows how to merge two shopping// carts. Last-write-wins would silently keep one and lose the other.function latest(versions: Versioned[]): Versioned[] { const out: Versioned[] = []; for (const v of versions) { const obsolete = versions.some((o) => o.clock.descends(v.clock) && !v.clock.descends(o.clock)); // Equal clocks are the same version read from two replicas: keep one copy. const duplicate = out.some((o) => o.clock.descends(v.clock) && v.clock.descends(o.clock)); if (!obsolete && !duplicate) out.push(v); } return out;}
// The client writes back with the merged context of everything it read, so the new version// supersedes every sibling. Clocks grow by one entry per coordinator; Dynamo trimmed the// oldest entries past 10, which can rarely turn an ancestor into a false sibling.function write(value: string, readContext: Versioned[], coordinator: string): Versioned { const ctx = readContext.reduce((acc, v) => acc.merge(v.clock), new VClock()); return { value, clock: ctx.tick(coordinator) };}Go deeperTwo traps with clashing writes1 min
There’s a catch with merging carts by union. A union can’t express “removed”. If one user removed the pen, the merge adds it back. Dynamo’s cart accepted that deleted items could reappear. A smarter set that keeps removal markers inside the value can get this right.
Adding nodes without trouble
Section titled “Adding nodes without trouble”We need to grow from 70 to 90 nodes before peak season. What happens when one node is added?
Consistent hashing limits how much data moves: about 1/71. It says nothing about moving it while those keys keep taking writes. So a new node joins in steps:
Adding a new node
1. Pick tokens
16 tokens, placed to even out the load across racks. Random tokens leave some nodes 30% fuller than others.
2. Join
Gossip announces the new tokens as “joining”. Writes for those slices now also go to the new node, and need one extra confirmation. Reads still go only to the current copies.
3. Copy
The node copies its slices from the current owners, slowly enough (100 to 200 MB a second) that they keep serving traffic. 1.2 TB takes 2 to 3 hours.
4. Serve, then clean up
The node starts serving reads, and the old owners stop being copies for those slices. Each then deletes data it no longer owns, as a separate step, so a failed join can be abandoned safely.
Two joins at once could change the same copy sets together, and break the quorum overlap from the first stress test. So ring changes go one at a time. At 2 to 3 hours each, 20 nodes is a week of work. Growth is planned a quarter ahead, never during the peak.
When things break
Section titled “When things break”One storage node is down for an hour at peak. What happens to the writes it misses?
Whether you know other nodes keep its writes as hints and hand them over when it returns.
No single machine, disk or rack should make any key unavailable at QUORUM.
A node is down for an hour. Every key it holds still has 2 live copies, so QUORUM reads and writes carry on. Each write it misses is kept as a hint by the coordinator that took it. When the node comes back, the hints are replayed to it, and it catches up in minutes. Reads in the meantime are still correct, because a QUORUM read always reaches at least one copy that has the newest write. If the node stays down past the 3-hour hint window, hints are dropped and repair copies the difference instead.
A whole rack loses power. Because every key has one copy per rack, a whole rack can go dark and every key still has 2 of its 3 copies. QUORUM requests carry on.
The cost arrives later. The other two racks take 50% more load, which is why nodes run at 60%. And a third of all writes become hints, about 100,000 a second. After an hour, it’s often cheaper to let hints expire and run repair instead.
Click any box to break it and see what users would notice:
Break it: what fails, who notices, what users see
Failure: click any box to break it
A storage node goes down
What breaks
Its replicas of about 4% of the keys stop answering.
Every key still has 2 live replicas.
How we notice
Gossip failure detector within about 10 seconds; node-down alert.
What users see
Nothing at QUORUM: 2 of 3 replicas still answer.
Requests at ALL fail for the keys it holds.
What we do
Coordinators store hints for it.
If it is gone for good, a replacement takes its tokens and streams its data from the other replicas, 2 to 3 hours for 1.2 TB.
One datacenter is still one failure: a four-hour outage spends years of a 99.99% budget. So we run a second datacenter with a full ring, and 3 copies of every key in each. Clients use LOCAL_QUORUM: 2 of the 3 copies in their own datacenter. Writes cross to the other datacenter in the background, in tens of milliseconds, so the 80 ms between them is never on the request path.
Stage 6: A second datacenter
Stage 6 of 6
A second datacenter
Why this stage
One datacenter is one failure domain: a single four hour outage spends more than four years of a 99.99% budget.
Users on another continent also pay 80 ms per round trip.
What we add
A full copy of the ring in a second datacenter, 3 replicas per DC, LOCAL_QUORUM reads and writes, and asynchronous replication between DCs.
What it costs
Twice the hardware, and concurrent writes in two DCs can conflict, so conflict handling is no longer optional.
At 600k reads per second, every part has headroom. Raise the traffic until something turns red.
The numbers behind the busy bars
What one machine can do
- One in-memory server handles about 500,000 reads a second, but only about 100,000 durable writes, because it waits on fsync for its log.
- One LSM node (16 cores, NVMe) does about 40,000 reads a second that touch disk and 60,000 writes.
- Mixed, that is about 50,000 replica operations a second, including the work of coordinating requests. The coordinator is a role storage nodes play, not separate servers.
Why replicas add work
- From stage 4, each client write becomes 3 replica writes, and each quorum read becomes 2 replica reads.
- From stage 6, each data center applies every write, and is sized to take all reads if the other one fails.
- Instance counts keep each box near 60% busy.
With two active datacenters, two writes to one key at the same moment are now routine. Both succeed locally and meet later. That makes the clash policy from the stress tests a requirement, not an option.
Go deeperEvery failure, and what users see2 min
| Part | Failure | What users see | How we notice | What we do |
|---|---|---|---|---|
| Storage node | Crash or hang | Nothing at QUORUM; ALL fails for its keys | Failure detector, about 10 s | Hints; replace and copy if it’s gone |
| Whole rack | Power or switch | Nothing at QUORUM: one copy per rack | A whole rack down at once | Other racks take 50% more; hints, then repair |
| Compaction | Falls behind | Reads slow down over hours | Files per read above 4 | Slow writes on that node; add nodes |
| Repair | Not running | Nothing for days, then deleted data returns | Days since last repair | Alert at 7 days; repair can resume |
| Network split | Racks or DCs cut off | Keys with 2 copies across the cut fail at QUORUM | Unreachable-node counts | On heal: hints, read repair, clash policy |
| Datacenter | Outage | Clients move; recent writes at risk | External test requests | The other DC serves all; repair before it rejoins |
The bigger risk is failures that happen together. One node in 70 failing is routine. A bad deploy on every node at once is not. So deploys go one rack at a time, and disks stay under half full.
Replication protects against hardware loss, not against a buggy client deleting data. Copies copy mistakes faithfully. SSTables never change, so a backup is a set of file links taken in seconds, then copied to object storage. Test a full restore every quarter, not during an incident.
Go deeperWhat each operation promises1 min
| Operation | Promise | What a user can see |
|---|---|---|
| QUORUM write, then QUORUM read | The read sees the write (strict quorums) | Nothing odd while healthy |
| Read at ONE | Eventual | A value a few ms old; older after failures |
Write that returned 504 |
Unknown: on 0 to 3 copies | A “failed” write showing up later |
| Clashing writes, timestamp table | One survives | A silently lost update |
| Clashing writes, sibling table | Both kept | The application sees siblings and merges |
| Across datacenters | Eventual, usually under 100 ms | A user moved to the other DC briefly reads older data |
| Delete | Holds only if repair runs within the grace period | A deleted key returning, if repair is neglected |
Go deeperRunning it: alerts, deploys and tenants1 min
Wake someone up when: fewer than 99.99% of QUORUM requests succeed for 5 minutes; the slowest 1% take over 10 ms for 10 minutes; reads check more than 4 files (this predicts slow reads hours ahead); a slice hasn’t been repaired for 9 days; or hints keep piling up, which means a node is flapping.
Deploys. Restart one rack at a time, a few nodes at once. First flush the memtable so the restart replays nothing. Wait until hints are replayed, then move on. A rack holds at most one copy of any key, so QUORUM never notices.
Tenants. Quotas cap operations, bytes and value sizes per tenant. The usual troublemakers are huge values, millions of tombstones that later reads must step over, and hot keys.
Wrap up
Section titled “Wrap up”The whole design
Section titled “The whole design”Here is everything we built, in the order we built it. Use the buttons to step through it.
The full design, stage by stage
Stage 6 of 6
A second datacenter
Why this stage
One datacenter is one failure domain: a single four hour outage spends more than four years of a 99.99% budget.
Users on another continent also pay 80 ms per round trip.
What we add
A full copy of the ring in a second datacenter, 3 replicas per DC, LOCAL_QUORUM reads and writes, and asynchronous replication between DCs.
What it costs
Twice the hardware, and concurrent writes in two DCs can conflict, so conflict handling is no longer optional.
At 600k reads per second, every part has headroom. Raise the traffic until something turns red.
The numbers behind the busy bars
What one machine can do
- One in-memory server handles about 500,000 reads a second, but only about 100,000 durable writes, because it waits on fsync for its log.
- One LSM node (16 cores, NVMe) does about 40,000 reads a second that touch disk and 60,000 writes.
- Mixed, that is about 50,000 replica operations a second, including the work of coordinating requests. The coordinator is a role storage nodes play, not separate servers.
Why replicas add work
- From stage 4, each client write becomes 3 replica writes, and each quorum read becomes 2 replica reads.
- From stage 6, each data center applies every write, and is sized to take all reads if the other one fails.
- Instance counts keep each box near 60% busy.
The decisions, and what each one costs
Section titled “The decisions, and what each one costs”| Decision | We chose | We gave up | We’d choose differently if |
|---|---|---|---|
| Storage engine | LSM tree, leveled compaction | One-file reads; 10× rewriting | Reads dominated (B-tree) or keys fit RAM (Bitcask) |
| Splitting keys | Consistent hashing, 16 virtual nodes | Range scans | Operators wanted exact moves (fixed slots + a map) |
| Routing | Client library that knows the ring | Thin clients | Many languages lacked a library (proxy tier) |
| Copies | No leader, N = 3, quorums per request | Linearizability; clash-free writes | Callers needed strong consistency (leader per group) |
| Quorums | Strict; hints don’t count | Writes when most copies are unreachable | Writes had to always succeed (sloppy, like Dynamo) |
| Clashes | Timestamp wins by default, siblings per table | Zero client work for sibling tables | All data were blind overwrites |
| Repair | Hints, read repair, incremental Merkle repair | Disk and network for repair | Data were written once and never deleted |
| Datacenters | Both active, LOCAL_QUORUM | Recent writes if a DC dies | Writes had to survive a DC loss (EACH_QUORUM, 80 ms) |
How to spend 45 minutes
Section titled “How to spend 45 minutes”- 0 to 4. Data size, rates, whether writes must survive failures, compare-and-set.
- 4 to 9. 86 TB on disk, 2.1 million operations on copies a second, about 70 nodes, 1.2 TB each, a 2 to 3 hour rebuild.
- 9 to 22. Build it: one node’s LSM tree, the ring, quorums, then repair. A mid-level answer can stop before repair.
- 22 to 33. The stress tests: what quorums promise, then clashing writes.
- 33 to 41. Adding nodes, or more on repair, whichever the interviewer picks.
- 41 to 45. A rack failure, then your own questions.
Never skip what R + W > N does not promise. If time is short, shrink the LSM tree to three sentences and drop the second datacenter.
Follow-up questions
Section titled “Follow-up questions”Interviewer asksOne key is taking 200,000 reads a second. What do you do?
Its 3 local copies serve about 150,000 operations a second between them, so no quorum setting saves it. First ask why. A hot key is often a design bug, like a config blob every request reads. Then cache it on coordinators for a second or two. Past that, the caller stores it under several names and reads one at random.
Interviewer asksA caller needs a counter that never loses an increment. What do you offer?
Not read-change-write over quorums, which loses increments that happen together. Either compare-and-set through Paxos, correct at about 4 round trips per update. Or a counter CRDT, where each copy counts its own increments and the value is the sum. That’s fast and correct, but it can’t enforce “never below zero”.
Go deeperMore follow-up questions2 min
How does TTL work with an LSM tree? Each value stores its expiry time. A read treats an expired value as a tombstone, so correctness never waits for cleanup. Compaction drops it after the grace period, like any delete. When all data in a table expires on a schedule, compaction can group values by write time, so whole files expire together and are deleted without being rewritten.
A user writes in Europe, then reads old data from the US. Is that a bug? It’s the promise we chose: LOCAL_QUORUM is consistent within a datacenter and eventual across them, usually within 100 ms. If it matters, keep the user’s session in one datacenter.
Version vectors grow with every coordinator. How do you keep them small? Dynamo dropped the oldest entry past 10, which can rarely create a needless sibling. Better: only copies of the key coordinate (the client library already sends to one), so a vector has at most 3 entries per datacenter.
Further reading
Section titled “Further reading”- Consistent hashing, replication, consistency models and CAP and PACELC on this site, and the URL shortener for a fixed slot map in practice.
- Dynamo: Amazon’s Highly Available Key-value Store, SOSP 2007: consistent hashing, sloppy quorums, hinted handoff, vector clocks and Merkle trees.
- Amazon DynamoDB: A Scalable, Predictably Performant, and Fully Managed NoSQL Database Service, USENIX ATC 2022: why the managed successor uses a leader per partition.
The designbeyondcode digest, every Saturday
New deep dives, interview tips and site news in one short email. Free. Leave with one click.
We never sell or share your email. We only use it to send this newsletter.