How to design a distributed key-value store

A distributed key-value store is a database with two operations, along the lines of Amazon’s Dynamo or Cassandra. Hand it a key and a value and it stores them. Hand it a key and it returns the value. No joins, no sorting, no searching by anything except the key.

In this lesson, we’ll take that small interface and derive the distributed system behind it: how many machines hold each copy, what happens when those copies disagree, and the mechanisms that notice the disagreement, bound it, and repair it.

By the end you’ll be able to explain:

• what a quorum buys and what it does not;
• what version metadata costs per record;
• how you find one wrong key among a million by exchanging about a kilobyte;
• why a bloom filter is what makes this storage engine viable, not an optimization on top of it;
• what “always available” costs, measured in stale reads per day.

The interface

There are three calls. The return type of two of them is the design argument the rest of the lesson defends:

put("cart:42", b"{milk}", context)  ->  an acknowledgement, and a NEW context
get("cart:42")                      ->  a LIST of values, and a context
delete("cart:42", context)          ->  an acknowledgement

Three terms recur throughout, defined here.

A get returns a list of values, not one value, because two clients can write the same key at the same instant on two different machines, and this store keeps both instead of guessing a winner. Two values returned side by side like that are called siblings.

A context is a small opaque token the store returns with every read. You hand it back on the next put to that key, and it is how the store knows which version you were looking at when you decided to write.

The rest of the lesson is the machinery that makes those three calls work at 100,000 writes a second across 16 machines.

The one prerequisite: consistent hashing

The store decides which machines hold a key by consistent hashing: hash the key to a point on a circle of hash values, walk clockwise from that point, and the first N distinct machines you meet own the key.

That ordered list of N machines is the key’s preference list, and N is the replication factor: the number of copies kept, three here. Arranging the hash values in a circle means adding or removing a machine moves only the keys in one arc, not every key. The consistent-hashing chapter derives that in full; here you need one sentence of it: preference_list(key) returns three machines, and those three machines are allowed to disagree.

Storage-engine vocabulary

A handful of terms recur throughout, each derived at length in the database-internals chapter and used here as a result.

• LSM tree (log-structured merge tree): a storage engine that only ever appends. Writes go to memory, memory is flushed to immutable files, and those files are merged in the background. Contrast a B-tree, which updates pages in place.
• WAL (write-ahead log): an append-only file that every change is written to before anything else happens, so a crash can be replayed from it.
• Write amplification: how many bytes the disk actually writes for each byte the application asked to write. Read amplification is its mirror: how many separate places the engine must look to answer one read.
• IOPS: how many individual reads or writes a storage device can service each second. For a random-access workload this runs out long before the device’s bytes-per-second figure does.
• Tombstone: a stored marker meaning “this key was deleted,” written instead of erasing the row. In a replicated store an absent value and a deleted value are otherwise indistinguishable, and a lagging replica would resurrect the data.

What decides everything: who resolves disagreement

A key-value store makes one central decision, and it is not the data structure. It is: when the N copies of a key disagree, who notices, and who resolves it? A replica is one machine’s copy of a key; N = 3 here, so every key exists three times.

Every real difference between the well-known stores is a different answer to that one question. Dynamo hands the conflict back and lets the application resolve it. Cassandra picks a winner by timestamp. Google’s Spanner refuses to let the copies diverge in the first place, at the cost of a coordination round trip on every write.

The other two problems are already solved before this lesson starts. Placing keys is settled by consistent hashing. Surviving one dead machine is settled by keeping three copies. Replicas disagreeing, on the other hand, happens continuously, and it is invisible.

The failure that matters here is not downtime but silent wrong answers. A store that returns a stale value quickly and confidently is worse than one that returns an error, because the error surfaces immediately while the stale read may not be noticed for weeks.

A store is eventually consistent when its replicas are permitted to disagree for a while and are guaranteed to converge only if updates stop. That is the entire content of the guarantee, and what it leaves out matters as much as what it states: it fixes no bound on how long the disagreement lasts, no bound on how stale a read can be, and nothing about what happens under continuous writes, which is the only regime a production store is ever in. Contrast strong consistency, where every read reflects every completed write and the copies are never observably different.

Wherever possible this lesson replaces the phrase “eventually consistent” with a count: 26,000 exposed writes a day, 900 sibling keys per 60-second partition, a 3-hour hint window, a weekly repair. Each number that replaces it is the actual answer.

Requirements

Functional

• get(key) -> (versions, context): note the plural. A get may legitimately return more than one value, and a caller that cannot handle that has a latent bug.
• put(key, value, context): the context is not optional (see vector clocks below). delete(key, context) writes a tombstone instead of erasing the row.
• W and R are tunable per request. W is how many of the three copies must acknowledge before a write is called done; R is how many must answer before a read is called done. Different callers want different points on the same trade, so it is a per-call argument, not a cluster setting.
• No range scans, no secondary indexes, no transactions spanning more than one key. Giving these up is what pays for everything else.

Non-functional

p99 below is the 99th percentile: the latency 99 of every 100 requests come in under.

Back-of-envelope

Two premises drive every figure: 500 million users, 20 stored objects each, giving 10 billion objects. (Twenty objects each is a shopping cart, a profile, and some session state.) The consistent-hashing chapter fixes S = 16 servers, RF 3, and 100,000 writes per second, reused here unchanged. The 100,000 reads per second is this lesson’s own assumption, and it sizes the bloom-filter argument later.

One notation collision: the Dynamo literature calls the replica count N, while the consistent-hashing chapter calls the fleet size N. Here the fleet is S = 16 and the replica set is N = 3.

Each record is 1,088 bytes: a 32 B key, a 1,024 B value, an 8 B timestamp, and a 24 B vector-clock entry. From there the sizing is order-of-magnitude arithmetic:

• Logical corpus: 10 billion objects x 1,088 B ≈ 10.88 TB, or 32.64 TB replicated three times, so 2.04 TB per node and 1.875 billion keys per node.
• Write bandwidth: every client write becomes three physical writes, so 300,000 replica writes/s across the fleet, 18,750/s per node, or 20.4 MB/s. A 1 Gbps NIC carries 125 MB/s, so the steady write path alone is 16% of the network card.

That last figure is the important one. The other 84% of the card is the entire budget for repair traffic, replaying writes parked elsewhere while a node was down, and moving data when the fleet changes size, all of which fire at once during an incident. That is why the scaling section is about bandwidth, not CPU. The 1.875 billion keys per node is what sizes bloom-filter RAM later.

API sketch

The type annotations carry the design:

class KeyValueStore:
    def get(self, key: bytes, r: int = 2) -> tuple[list[bytes], bytes]:
        """(siblings, context). len(siblings) > 1 means unresolved conflict."""
    def put(self, key: bytes, value: bytes, context: bytes, w: int = 2) -> bytes:
        """context comes from a prior get. Returns the new context."""
    def delete(self, key: bytes, context: bytes, w: int = 2) -> None:
        """Writes a tombstone. Not gone until compaction says so."""

Two details are load-bearing. The first is that get returns a list, which forces every caller to handle conflicting values when they write the code, not discover the problem in production. The second is that there is no put without a context: the only way to write without one is to pass an empty context, which is an explicit statement of “I accept that my write will become a sibling,” not an accident.

Data model

Keys are opaque bytes: the store never looks inside them, it only hashes them, which is what forecloses ordering and range scans. Values are opaque blobs capped at 1 MB; anything larger goes to a separate blob store with the value holding only the URL.

The 1,088 bytes per record breaks down as:

key                                              32 B
value                                         1,024 B
timestamp                                         8 B
vector clock, 1 entry x 24 B                     24 B
                                              -------
record                                        1,088 B

A clock entry is the triple (node id 8 B, counter 8 B, timestamp 8 B) (24 B), and a record carries exactly one entry in the common case. What happens when it carries more, and what that costs, is covered in the vector-clock section.

High-level architecture

Every node runs identical software, and any node can act as coordinator for any request. The coordinator is simply whichever node the client happened to connect to. Its job is to forward the request to the three replicas and count the answers.

There is no master, no configuration server, and no separate tier storing where the data lives. That is possible because the ring is a pure function of the membership set: give any node the list of live machines and it computes the same preference list as every other node.

The diagram below has two halves. The top is the fleet-level path a request takes; the boxed subgraph is what happens inside one replica once the request lands there.

The client reaches any coordinator, which computes the preference list, gets three distinct machines, and sends the request to all three. It then checks the condition the design turns on: has a write collected W acknowledgements, or a read collected R answers? If so, it returns a bare acknowledgement for a write, or the list of sibling values plus a context token for a read. If a replica is unreachable, the coordinator falls back to a sloppy quorum: it sends the write to the next healthy machine, which stores a hint. Running underneath, gossip with phi-accrual failure detection keeps the coordinator’s idea of the membership set current.

The boxed subgraph is the storage engine, running in four steps: append to the WAL and fsync; insert into the memtable (a sorted table in RAM); flush a full memtable to an immutable 64 MB SSTable (a sorted string table, written in key order and never modified); and compaction, the background job that merges those files. Two side structures hang off it, off the write path: a bloom filter per SSTable that answers “is this key definitely absent from this file?” out of RAM, and a Merkle tree per key range used for anti-entropy, repairing a replica by comparing it with its peers.

flowchart TB
    C(["client"]) --> CO["coordinator · any node"]
    CO --> PL["preference list<br/>3 distinct nodes"]
    PL --> R1["replica s3"] --> Q{"W acks?<br/>R responses?"}
    PL --> R2["replica s7"] --> Q
    PL --> R3["replica s11"] --> Q
    Q -->|"yes"| OK(["ack / siblings + context"])
    Q -->|"no, node down"| HH["sloppy quorum<br/>next healthy node<br/>stores a HINT"]

    subgraph REP["inside one replica"]
        W["1 WAL append + fsync"] --> M["2 memtable · sorted, RAM"]
        M --> S["3 flush -> immutable<br/>SSTable, 64 MB"] --> K["4 compaction"]
        B["bloom filter<br/>per SSTable"] -.-> S
        MT["Merkle tree<br/>per key range"] -.-> AE["anti-entropy<br/>repair with peers"]
    end

    R1 --> REP
    G["gossip + phi-accrual"] -.-> CO

Two properties of this storage engine matter. The authoritative bytes live only in the memtable and the SSTables, not in the Merkle tree, which holds only hashes and can be rebuilt without losing a byte. And the two side structures exist to keep work off the request path: the bloom filter turns eight disk reads into 0.066, and the Merkle tree turns 1.14 GB of streaming into about a kilobyte, both derived below.

The mechanisms

Seven, in the order a request meets them:

1. The quorum: how many copies a call touches.
2. Vector clocks: the version metadata that detects disagreement.
3. Merkle trees: the hash tree that finds disagreement cheaply.
4. The storage engine: what actually serves the bytes, and the bloom filter.
5. Hinted handoff: the fallback that keeps writes flowing when a replica is down.
6. Phi-accrual: how the cluster decides a replica is down.
7. CAP: what all of it adds up to during a network partition.

1. The quorum: what W and R buy

A quorum is a minimum number of copies that must participate in an operation. The rule is W + R > N, where W is how many replicas must acknowledge a write and R how many must answer a read.

The reason is the pigeonhole principle. The write landed on some W of the N replicas and the read consults some R of them. Two subsets of an N-element set must share at least W + R - N members, and W + R > N makes that at least 1. At N = 3, W = R = 2 the guaranteed overlap is 2 + 2 - 3 = 1, so at least one replica in the read set holds the write and the read cannot miss it.

One convention flips the intuition. The coordinator sends the request to all N replicas and returns as soon as the first W (or R) answer. So R = 1 means the fastest of three, not one node picked at random. A smaller W is faster not because it does less work, but because it discards the stragglers.

What W costs in latency. Suppose one replica exceeds 10 ms with probability p = 0.01, a realistic tail for one node doing a memtable insert plus a WAL fsync. The w-th response is slow only when at least 3 - w + 1 replicas were slow. Counting those cases:

• W = 1, slow only if all three were slow: 0.01³ = 0.0001%.
• W = 2, slow if two or three were: 0.0298%.
• W = 3, slow if any one was: 1 - 0.99³ = 2.97%.

Waiting for 2 of 3 gives a tail 34x better than a single replica, because you discard the slowest of three. Waiting for all 3 gives one 3x worse. A quorum is not a tax on latency; only a full quorum is. Waiting for 2 of 3 is a free hedged request: issue redundant copies and take the first answer.

What W costs in availability. Assume each node is up 99.9% of the time, about 8.8 hours of downtime a year from deploys, restarts, and garbage-collection pauses, not disasters. A write is blocked when fewer than W replicas are up:

• W = 3 blocked: 1 - 0.999³ = 0.2997%.
• W = 2 blocked: 0.0003%.

W = 3 is a thousand times less available than W = 2, and buys nothing W = 2, R = 2 does not already give.

The full menu at N = 3:

Compare (2,2) against (3,1) in both directions: on writes (2,2) is about 100x better, on reads it is 298x worse. Neither dominates. (3,1) moves the whole cost of the third replica onto the write path and buys a read path as fast as one node; (2,2) splits it evenly. The choice is which of the two paths absorbs the third replica’s tail latency, not fast versus slow.

The four assumptions under all of it. Each fails in a specific way, and this is where the arithmetic meets a real cluster.

1. Replicas are slow independently. Multiplying per-replica probabilities is only legal if one replica being slow tells you nothing about the others. In a real fleet it often tells you everything: a rolling deploy, a shared top-of-rack switch, a compaction storm, or a coordinated GC pause hits all three at once. When failures correlate, the 0.2997% figure collapses toward the single-node 0.001, and the thousand-fold advantage of W = 2 shrinks toward nothing. The mitigation is rack-aware placement.
2. p = 0.01 and 99.9% are stable inputs. They are measurements that move with load. Treat the table as a shape (full quorum is roughly two orders worse than partial quorum), not as five decimal places.
3. The read set and write set are drawn from the same N replicas. This is the load-bearing one: the pigeonhole argument is about two subsets of one set. The moment a write lands on a machine outside the preference list, the two sets can be disjoint no matter what W + R sums to. That is exactly what sloppy quorum does, quantified later at about 26,000 writes a day.
4. “The last completed write” is well-defined. It is not, when two clients write at the same instant. The overlap guarantee says the read set contains the last completed write; it says nothing about which of two concurrent writes should win, which is why vector clocks exist.

What (1, 1) gives up is countable, not abstract. The write is acknowledged as soon as the fastest replica has it, while the other two are still in flight, the replication window. A read inside that window lands on the replica that already has the value with probability at least 1/3 (the figure for a random replica; in practice higher, because R = 1 takes the fastest answer and the replica that acked first tends to answer first). The honest statement is the bound: at least a third of reads inside the replication window are fresh, so up to two-thirds return the previous value. The window is normally under a millisecond, but it has no upper bound during a GC pause, a partition, or a slow disk, and nothing in the protocol reports it. (1,1) is “correct except when something is wrong.”

Quorum does not give linearizability. Linearizability is the strongest single-object guarantee: every operation appears to take effect at one instant between its call and return, so once a read returns a new value no later read returns an older one. Quorum overlap gives only “the read set intersects the replicas that acked the last completed write.” Four gaps remain: a write still in flight can make two reads see old, then new, then old again; a write that reached one replica and then failed is never rolled back; concurrent writes still need vector clocks to order them; and sloppy quorum breaks the intersection outright.

Two weaker guarantees are achievable. Read-your-writes (a client sees its own last write) holds if that client’s reads and writes go through one coordinator. Monotonic reads (never seeing time run backwards) holds only once you add read repair: the coordinator noticing, while already holding the replicas’ answers to a read, that some are behind, and writing the current value back to them on the spot. It costs no extra requests. A truly linearizable register needs consensus: Raft or Paxos, where a majority agrees on one ordered log before any operation is applied.

2. Vector clocks, and the sibling nobody wants

The quorum guarantees the read set contains the last completed write, but it cannot order two concurrent writes. The store detects that two writes conflict without deciding which wins.

A vector clock is a small map from node name to counter, carried alongside every version of a value. Each time a coordinator accepts a write for a key, it increments its own counter in that key’s clock, so the clock records how many writes each machine has contributed to this key.

Clock A descends from clock B when A[n] >= B[n] for every node n in B: whoever produced A had already seen everything B contains. If neither descends from the other, neither writer had seen the other’s write. The two writes are concurrent, and both must be kept as siblings. The clock detects concurrency exactly, and never decides a winner.

The diagram below is the whole mechanism in eight messages. Two clients write key cart:42 through two different coordinators, Sx and Sy. Both machines are healthy. Concurrency needs nothing to go wrong.

sequenceDiagram
    participant A as client A
    participant Sx as coordinator Sx
    participant Sy as coordinator Sy
    participant B as client B
    A->>Sx: put cart:42 = {milk}, empty context
    Sx-->>A: D1 clock [(Sx,1)]
    B->>Sy: put cart:42 = {eggs}, empty context
    Sy-->>B: D2 clock [(Sy,1)]
    Note over Sx,Sy: D1 has Sx=1 > 0, D2 has Sy=1 > 0<br/>neither descends -- CONCURRENT
    A->>Sx: get cart:42
    Sx-->>A: siblings {milk} and {eggs}<br/>context [(Sx,1),(Sy,1)]
    Note over A: application merges: a cart is a UNION
    A->>Sx: put {milk,eggs} with that context
    Sx-->>A: D3 clock [(Sx,2),(Sy,1)]
    Note over Sx,Sy: D3 descends from both -- siblings collapse

The merged context [(Sx,1),(Sy,1)] is the pointwise maximum of the siblings’ clocks: for each node name, keep the larger counter. When the client writes back with it, Sx increments its own entry to give [(Sx,2),(Sy,1)], which descends from both siblings, so both are discarded and one value remains.

The store never resolved anything. It kept the conflict visible until an application that knows a shopping cart is a set of items decided the merge. The database’s job is to lose neither write and refuse to guess between them.

Four subtleties decide whether an implementation is correct:

1. Deduplicate on the pair (clock, value), not the clock alone. Two writes can carry an identical clock and different values: a blind write, a put not preceded by a get, so both clients started from the same context. When the clocks are equal, no clock comparison can tell the two values apart, so read repair finds nothing and both values must be kept. Deduplicating on the clock alone would discard a real write with no error and no trace.
2. Read repair covers every replica that answered, not just the first R. A response arriving after the read returned is free information, and a replica holding nothing at all for the key is both the most stale and the least likely to be among the first R. Scoping repair to the quorum leaves known-stale replicas known-stale.
3. In the common case the repair set is empty. A replica that already holds a descendant of every sibling is not written back to. All three agreeing costs nothing.
4. An empty clock is never treated as an ancestor. descends(anything, {}) is vacuously true, so left unguarded an empty-context write would be an ancestor of every version and dropped silently, the exact loss the API promises an empty context does not cause. A production coordinator stamps its own (node, counter) onto a write before storing it, so the guarantee lives in the coordinator. Exactly one of the coordinator or the merge rule must carry it.

The clock grows without bound. A key’s clock gains one entry per distinct coordinator that has ever written it. Under normal routing that is one entry; under coordinator failover, sloppy quorum, or a client that reconnects to a different node each time, it grows toward the fleet size, 16 here, 1,000 in a large cluster. A 10-entry clock is 240 B, which is +19.9% on a 1,088 B record and 23% of the 1,024 B payload it describes. At 1.875 billion keys per node that is real disk, network, and CPU.

The standard fix can lose data. Dynamo caps the clock at 10 entries and evicts the oldest by timestamp. The harmless case is a descendant that loses an entry its ancestor still has, producing a spurious but safe sibling. The dangerous case is not: take two genuinely concurrent writes {S1:1, S2:1} and {S1:1, S3:1}. Truncate the first to {S1:1}, and now the second descends from it: the store treats one concurrent write as an ancestor of the other and discards it, with no sibling and no error. It almost never fires (Dynamo reported 99.94% of requests over 24 hours saw exactly one version), but “almost never” plus “undetectable” is the profile of a bug that takes a year to find.

Three ways out:

1. Keep the clock server-side and per-replica, not per-coordinator. The entry count is then bounded by N = 3, not by the fleet. This is the version vector, and it makes the growth problem mostly theoretical in a well-routed system.
2. Dotted version vectors (DVVs). Plain version vectors still accumulate siblings when one client writes repeatedly without reading in between. A DVV attaches a dot (a single (node, counter) identifying one specific write) next to the summarizing vector, letting the store tell a fresh write from a client that had seen version 5 apart from a genuinely concurrent write, so repeated blind writes stop manufacturing siblings.
3. Last-write-wins (LWW), Cassandra’s default. No metadata, no client reconciliation. The cost is exact: of two concurrent writes, one is discarded and the wall clock picks which. Machine clocks are kept in step by NTP to within about 10 ms in a datacenter, so two writes 2 ms apart can be ordered backwards, and the loser leaves no trace. LWW is correct for immutable or idempotent data; it is wrong for a shopping cart, a counter, or anything a human will later ask “where did my edit go?” about.

3. Merkle trees: one bad key in a million

Two replicas hold the same range of keys. Are they identical? Streaming the whole range across to compare works but is unaffordable.

A Merkle tree (hash tree) is the efficient answer. It is a binary tree: every leaf is the hash of one key’s (key, version) pair, and every internal node is the hash of its two children concatenated. A hash here is a short fixed-length fingerprint (32 bytes) that changes if any input byte changes. The consequence is the point: the root is a 32-byte fingerprint of the entire range.

Take a range of 2²⁰ = 1,048,576 keys. Streaming it costs 1.14 GB and 9.13 seconds of a saturated network card, per range, per pair of peers. Even shipping one 32-byte digest per key costs 33.6 MB. Both grow in direct proportion to the key count, O(n), which is why nobody compares replicas either way. The tree over 2²⁰ leaves is 20 levels deep and has 2 x 1,048,576 - 1 = 2,097,151 nodes.

If the two roots match, 1,048,576 keys are proven identical by exchanging 32 bytes. That is the common case. When they differ, you descend: at each level you compare only the two children of a node already known to be bad, so the total is the root plus two hashes per level: 1 + 2 x 20 = 41 hashes, 1,312 bytes. That cost grows with the logarithm of the key count, O(log n): doubling the range adds two hashes, not a million.

Round trips are a separate question from hashes. Descending one level per message costs 21 messages, about 10.5 ms at a 0.5 ms intra-datacenter round trip, cheap in bytes, expensive in latency. Shipping the whole tree in one message costs 67 MB. The workable answer sits between: ship the top 11 levels (2,047 nodes, 65 KB) in one message, then the one surviving subtree in a second: 2 round trips, 131 KB, 8,708x less traffic than streaming. Scaled to a whole node’s 1,788 ranges, 57 KB of root hashes proves 2.04 TB agrees with a peer, a 35-million-to-one compression of the question.

Three costs make this a scheduled job, not a continuous one:

• Building the tree is O(n) and a full range read. 2.04 TB at 1 GB/s sequential is 34 minutes per node, 9.1 hours for the fleet: cheap to compare, expensive to build, which is why repair is a weekly scheduled job.
• Any write invalidates the path from that leaf to the root. Maintain it incrementally (20 hash updates per write, trivial CPU) or rebuild at repair time. Cassandra rebuilds.
• Real implementations cap the depth, and the cap costs over-streaming. Cassandra caps at 2¹⁵ = 32,768 leaves, so each leaf covers 32 keys and one differing key streams 34,816 bytes instead of 1,088, a 32x over-stream, fine. But at 100 million keys per range a leaf covers 3,052 keys and one bad key streams 3.3 MB, the classic Cassandra “repair streams far more than my drift” complaint. The fix is more, smaller ranges, not a deeper tree.

What the Merkle comparison assumes: both peers build over identical range boundaries (a topology change that redraws boundaries invalidates every tree, which is why topology changes must be serialized); both trees describe the same instant (live writes mid-comparison are handled by snapshotting); hash collisions are impossible in practice (true for 32 bytes); and the number of differing keys is small. If half the range differs, the descent visits most of the tree and streaming would have been cheaper, which is exactly the regime a node down for a day is in.

4. The storage engine and the bloom filter

Inside one replica, a write is three steps on the critical path and two in the background. On the critical path: append the change to the WAL and fsync so the bytes are on the physical device; insert the key into the memtable; acknowledge. That is why a write costs one sequential disk append, not a random one: nothing is searched for or updated in place. In the background, a full memtable is flushed to an SSTable, and compaction merges those files, dropping superseded versions and expired tombstones.

At 20.4 MB/s per node, a 64 MB memtable flushes every 3.1 seconds, about 27,500 SSTables per node per day. At that rate compaction is the dominant consumer of the device, not a background nicety.

Choosing a compaction strategy, in bandwidth. The two strategies arrange those files differently:

• Leveled compaction keeps levels each about ten times the size of the one above, holding non-overlapping files; a key lives in at most one file per level. It minimizes reads, pays heavily on writes.
• Size-tiered compaction merges files of similar size as they accumulate, leaving several overlapping files a key might be in. It minimizes writes, pays on reads.

This node’s 2.04 TB needs 5 leveled levels (each level 10x the last, starting from 64 MB; the fifth reaches 6.4 TB). The leveled write-amplification formula 1 (WAL) + 1 (L0 flush) + T x L gives 1 + 1 + 10 x 5 = 52, so leveled needs 20.4 MB/s x 52 ≈ 1,061 MB/s against a 1 GB/s NVMe device: it does not fit, before a single read is served. Size-tiered runs at about 4.5x write amplification, or 92 MB/s, 9% of the device. (An NVMe is a solid-state drive on the PCIe bus, the fast end of what a server has.) That is why a Dynamo-style store defaults to size-tiered while an embedded engine like RocksDB defaults to leveled: this workload is write-dominated, and disk space is cheaper than disk bandwidth. You pay for the choice in read amplification and in space amplification of about 2x, so provision about 4.08 TB of disk per node.

The read path, and why it needs a filter. Reading checks the memtable, then each SSTable, newest first. Because size-tiered leaves several overlapping SSTables (call it 8), that is 8 random disk reads per lookup, and 7 of every 8 are guaranteed misses because a key lives in at most one SSTable until compaction merges them.

A bloom filter fixes this. It is a small bit array in RAM, one per SSTable, answering one question: “is this key definitely not in this file?” You add a key by hashing it to k positions and setting those bits; you query by checking whether all k bits are set. Because bits are shared, a key never added can find all k bits already set by others, a false positive, a wasted disk read. It never produces a false negative, because an added key always has its bits set. So a “no” is a proof and the SSTable is skipped without touching disk; a “maybe” costs at most one wasted read.

The false-positive rate is (1 - e^(-kn/m))^k where m is bits, n is keys, and k is probes, minimized at k = (m/n) ln 2. Substituting the optimal k collapses it to p = 0.6185^(m/n):

Every 4.8 additional bits per key divides the false-positive rate by 10, which is also why nobody goes past about 16 bits: by then the disk reads you save are noise. At the chosen 10 bits per key the rate is 0.82% for 2.34 GB of RAM per node, under 4% of a 64 GB box.

The payoff is large. For a key absent from all 8 SSTables, expected disk reads drop from 8 to 8 x 0.0082 = 0.066, a 122x reduction, and 93.6% of absent-key lookups touch no disk at all. In hardware, a 100 MB/s random device does about 24,414 IOPS with a 4 KB page, and this node needs 18,750 lookups/s. Without filters that is 150,000 IOPS, or 6.1 devices per node just to serve reads. With them it is about 19,826 IOPS, 0.81 of one device. That is why calling bloom filters an optimization is wrong: they are what makes the LSM read path viable at all. Six devices per node means buying a different machine, not tuning the one you have.

Two limits: a bloom filter cannot answer range queries, only exact keys; and it cannot forget, so deleting a key cannot unset its bits and a filter’s false-positive rate only drifts upward as an SSTable ages.

What the bloom arithmetic assumes. A real cluster violates all four: the filter is sized for the keys it actually holds (over-fill it 5x and m/n falls to 2, a 38% false-positive rate); the hash positions are independent and uniform (a weak hash clusters bits; the Kirsch-Mitzenmacher trick, two hashes combined as h1 + i·h2, produces k well-spread positions from one hash with no asymptotic loss); the filter stays resident in RAM (if paged out, every lookup costs a disk read for the filter on top of the SSTable read, inverting the saving into a penalty); and the page cache is cold. That last is an honest caveat: the page cache absorbs most hot-key reads in steady state, so the 6.1-devices figure overstates it, but the filter is what makes the cold path survivable, and the cold path is exactly where you are after a restart, during a compaction storm, or once a large scan has evicted the cache.

Digest reads save on the network, not the disk: the coordinator asks one replica for the full value and the rest for a 32-byte hash, requesting full values only on disagreement. At R = 3 that is 1,152 bytes on the wire instead of 3,264, 65% off the read path’s network cost, for one extra round trip in the rare disagreeing case.

5. Hinted handoff and sloppy quorum

A strict quorum requires the W acknowledgements from the top N machines of the preference list. At N = 3, W = 2, one node down is fine; two down means the write fails. A sloppy quorum drops that: if s1 is unreachable, the coordinator walks past the top three to the next healthy machine, s4, which stores a hint (the value plus a note saying “this belongs to s1”) and replays it once s1 returns. The whole mechanism is hinted handoff.

The moment it fires, W + R > N stops being true. The timeline below produces a stale read with no rule broken. At t0, s1 and s2 are unreachable, so the write acked by s3 and s4 is all the write there is, with s4 holding a hint for s1. At t1 they return and s3 is slow, so a read answered by s1 and s2 satisfies R = 2, and is stale, because neither reader was in the writer set.

flowchart LR
    T2["t0 · s1, s2 unreachable<br/>write acked by s3 and s4<br/>s4 holds a HINT for s1"] --> T3["t1 · s1, s2 return<br/>s3 is now slow"]
    T3 --> T4["read answered by s1 and s2<br/>R = 2 satisfied"]
    T4 --> T5["STALE · neither reader<br/>was in the writer set"]

The writer set {s3, s4} and reader set {s1, s2} share no members. W + R = 4 > 3 still held and bought nothing, because the pigeonhole argument assumed both sets were drawn from the same N nodes. Sloppy quorum trades away the exact guarantee quorum existed to provide.

Two rates, a thousand apart. Attaching the wrong number to the wrong question is an easy mistake. A sloppy write is triggered by one preference node unreachable. An actual stale read needs two, because with W = R = 2 a writer set still holding two preference nodes cannot be disjoint from any two-node reader set. At 99.9% per node:

• P(≥ 1 of 3 down) = 0.2997%, so about 300 writes/s route outside the preference list, about 25.9 million a day, the number that sizes hint volume.
• P(≥ 2 of 3 down) = 0.0003%, so about 26,000 a day are in a state where a later quorum read can miss them, the correctness number.

Both are numbers, which is the point: you can decide whether 26,000 exposed writes a day is acceptable for your product, whereas the phrase “eventually consistent” gives you nothing to weigh.

Hint volume, and the storm it causes. For a one-node outage at 18,750 writes/s destined for it:

• A 10-minute outage accumulates 11.25 million hints, 12.24 GB, and 98 seconds of replay at line rate, on top of the 20.4 MB/s of live traffic the recovering node is already accepting.
• A 3-hour outage (Cassandra’s default hint window) gives 220 GB and 29 minutes of replay. No node buffers that for a peer, which is why the window exists.

That replay flood is the hint storm: the node returns, every peer floods it at once, it falls over again, and more hints accumulate behind it. The mitigations are to rate-limit replay and to rejoin the ring for writes before reads. Past the hint window, hints are dropped and the only path back to agreement is Merkle comparison. Hinted handoff and anti-entropy repair are the fast and slow paths of the same job, and the hint window is the boundary between them.

Read repair is the cheapest of the three mechanisms. When a quorum read finds replicas disagreeing, the coordinator writes the merged value back to the stale ones, every replica that answered, not just the R the read waited for. It costs no extra requests and fixes exactly the keys people are reading. That cuts both ways: real access follows a Zipf distribution, a small hot set absorbing most traffic, so read repair fixes the hot head and does nothing for the cold tail, which is what scheduled anti-entropy is for. And it is blind to identical-clock siblings by construction, so those must simply be kept.

What hinted handoff assumes: outages are short relative to the hint window (past it, a cluster not running repair never converges, silently); the stand-in has spare disk and bandwidth (during a correlated incident it does not); the recovering node can absorb the replay (the storm arithmetic shows it often cannot without rate limiting); and the failure detector is right (a flapping node accumulates hints it never needed), which is why failure detection is part of this mechanism.

6. Membership: gossip and phi-accrual

All of the above depends on knowing who is alive, with no central registry to ask. Gossip is the protocol: once a second, every node picks a few peers at random and exchanges its view of which machines are in the cluster and how recently each was heard from. Information spreads like a rumour, so the fleet converges in rounds proportional to O(log S), about four rounds at S = 16, ten at a thousand nodes.

The interesting part is the failure detector, because a flapping node (one alternating between healthy and unreachable) forces the cluster to move data out and back on every transition. A fixed timeout cannot win: short, and a GC pause evicts a healthy node; long, and a dead node keeps being sent writes. The right answer depends on what the network is doing at that moment.

The phi-accrual detector replaces the yes-or-no answer with a continuous suspicion level. Each node sends a heartbeat on a regular interval. The detector records the gaps, fits a distribution to them, and reports:

phi(t) = -log10 P(next heartbeat arrives later than t)

Read that as the negative base-10 log of the probability a healthy node would have stayed silent this long. A phi of 8 means such a silence would be a one-in-a-hundred-million event. For exponentially distributed gaps with a 1-second mean, phi(t) = t / ln(10), so phi = 8 fires after 18.4 seconds of silence, and, the property a fixed timeout cannot have, that threshold stretches automatically when the network slows, because the fitted distribution widens with the observed gaps.

Two consequences. The output is a suspicion level, so different subsystems pick their own thresholds: stop routing reads at phi = 4, wait for phi = 8 plus a long timeout for ring membership. And “down” must never automatically mean “removed from the ring,” because removing a node redistributes 1/S of the corpus, 2.04 TB of movement here. Decommissioning stays a human decision.

7. CAP, stated for this store

CAP stands for Consistency, Availability, and Partition tolerance, and it describes what a system does while a network partition is in progress, two groups of machines both running but unable to reach each other. It is not a menu you pick two items from. Partitions are imposed by the physical world, so P is not a choice; what you choose is how the reachable side behaves. Two corollaries: a single-node store is not “CA,” because it is not a distributed system and the theorem says nothing about it; and outside a partition the theorem constrains nothing.

PACELC covers that second case: during a Partition you trade Availability against Consistency; Else, when the network is fine, you trade Latency against Consistency.

What is specific to this store is that W is the knob, set per request. Suppose a partition splits the preference list into {s1, s2} and {s3}. In the classification column, PC means the system chose consistency over availability, PA the reverse:

Pricing W = 1: a 60-second partition at 100,000 writes/s with traffic split evenly is 3 million writes per side. The chance a given right-side write hits a specific left-side key is 1 / 10 billion, so the expected collisions are 3,000,000 x 3,000,000 / 10 billion = 900 keys written on both sides: 900 siblings out of 6 million writes, 0.015%, which lines up with Amazon’s reported 99.94% of Dynamo requests seeing exactly one version. That is the defence of W = 1: conflicts are rare because 10 billion keys is an enormous space. The defence assumes writes are spread uniformly, and fails under a Zipf distribution, where a hot set concentrates collisions and the count rises by orders of magnitude.

Bottlenecks and scaling

What runs out first depends on the operating regime. Two abbreviations: QPS is queries per second, and gc_grace is Cassandra’s setting for how long a tombstone is kept before compaction may discard it. Four of the seven regimes bind on bandwidth, not CPU or disk space.

Running across multiple regions is where the quorum choice stops being subtle. With three copies spread one per region, a W = 2 write must wait for a second region, so it cannot finish faster than one cross-region round trip, about 70 ms against 0.5 ms within a region, 140x the local cost, on the user’s critical path. The standard answer keeps three copies per region and uses LOCAL_QUORUM (count acknowledgements only from the caller’s own region), accepting that cross-region convergence is asynchronous and a regional partition produces siblings. This is the “Else” branch of PACELC, and it applies to nearly every request.

Failure modes

Each row is one way the design breaks in production. The theme is that the dangerous failures are the silent ones (a stale read, a truncated clock, a resurrected tombstone all produce wrong answers with no error anywhere), which is why the Detection column matters more than the Guard column.

Alternatives considered

Several other designs are reasonable. Each is genuinely good at something and ruled out here by a specific piece of arithmetic or a specific requirement. One term first: a CRDT (conflict-free replicated data type) is a data structure whose merge is defined so any two replicas that have seen the same set of updates, in any order, end up identical. A counter or a set can be built this way; an arbitrary blob cannot.

The one to revisit, with a trigger: move to per-key consensus the first time an application team asks “what should I do when I get two values back?” and has no answer. Sibling reconciliation is a product decision delegated to the client, and a client that cannot make it will write siblings[0] and silently lose data.

Conclusion

The whole design turns on one decision: when three copies of a key disagree, who resolves it? This store’s answer is to detect disagreement precisely and hand it back, never to guess. Everything else follows:

• W = 2, R = 2 is the default at N = 3. It guarantees a one-replica overlap and keeps both tails better than a single replica. W = 3 is a thousand times less available for nothing; (1,1) gives up the overlap and returns stale values with no error during any incident.
• Vector clocks detect concurrency but never resolve it. They cost about +20% per record, grow with the number of distinct coordinators, and the standard truncation fix can silently drop a concurrent write. Bound them with server-side version vectors and dotted version vectors.
• Merkle trees find one bad key in a million by exchanging about a kilobyte, but building the tree is a full sequential read, so repair is a scheduled weekly job.
• The bloom filter is what makes the LSM read path viable: it turns roughly 6 devices per node into less than one.
• Hinted handoff keeps writes flowing when a replica is down, at a known price: about 26,000 writes a day can be read stale, and past the 3-hour hint window Merkle repair is the only path back to convergence.
• CAP describes only behaviour during a partition, and here it is a per-request knob. W = 2 is PC, W = 1 is PA. The trade you actually pay on every request is latency versus consistency, the “Else” of PACELC.

Every one of those is a number you can hold your product’s requirements up against, which is more than “eventually consistent” ever tells you.

One line to remember: the whole store turns on one decision (when three copies disagree, detect it precisely and hand it back, never guess) and W = 2, R = 2 at N = 3 is the default that makes it work.

Further reading

• DeCandia et al., “Dynamo: Amazon’s Highly Available Key-value Store” (SOSP 2007): the source for the quorum, vector clocks, hinted handoff, and anti-entropy design used here.
• Lakshman and Malik, “Cassandra, A Decentralized Structured Storage System” (2010): the source for last-write-wins, size-tiered compaction, and gc_grace.
• Hayashibara et al., “The φ Accrual Failure Detector” (2004): the continuous-suspicion failure detector.
• Gilbert and Lynch, “Brewer’s Conjecture and the Feasibility of Consistent, Available, Partition-Tolerant Web Services” (2002): the formal proof of the CAP theorem.
• Bloom, “Space/Time Trade-offs in Hash Coding with Allowable Errors” (1970), and Kirsch and Mitzenmacher, “Less Hashing, Same Performance” (2006): the filter and the two-hash construction.
• Preguiça et al., “Dotted Version Vectors: Logical Clocks for Optimistic Replication” (2010): the fix for sibling accumulation under blind writes.

Next: Design a unique ID generator, where the key this store partitions by has to come from somewhere, and 64 bits is a budget you spend deliberately.