06 — Design A Key-Value Store

“Design a distributed key-value store. Something like Dynamo or Cassandra.”

A key-value store is a database with two operations and nothing else. Hand it a key and a value and it stores them. Hand it a key and it gives the value back. No joins, no sorting, no searching by anything except the key.

This chapter takes that deliberately tiny interface and derives the whole distributed system behind it: how many machines hold each copy, what happens when those copies disagree, and every mechanism that exists to notice the disagreement, put a bound on it, and eventually repair it.

When you finish you should be able to explain to someone else, in numbers rather than adjectives:

• 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 rather than a nice optimization on top of it;
• what “always available” costs, measured in stale reads per day.

What goes in and what comes out. There are three calls. The block below is the whole public interface — read the return side of each line, because the shape of two of them is the design argument the rest of the chapter 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 in those three lines, defined once and used everywhere below.

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 rather than 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.

Everything else in this chapter is the machinery that makes those three lines true at 100,000 writes a second across 16 machines.

The one prerequisite. 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. The point of arranging the hash values in a circle is that adding or removing a machine moves only the keys in one arc, rather than reshuffling every key in the system; chapter 05 derives that in full.

For this chapter you need exactly one sentence of it: preference_list(key) returns three machines, and this chapter is about the fact that those three machines are allowed to disagree.

The table below is the chapter in five rows. Each row is a question, the number this chapter answers it with, and where that number is derived. If you only remember five things, remember the middle column.

Storage-engine vocabulary, fixed up front. A handful of terms recur throughout this chapter. Each is derived at length in sql 03 and used here as a result, so pin the meanings now and you will not have to stop later.

• 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 (input/output operations per second) — how many individual reads or writes a storage device can service each second. For a random-access workload this is the number that 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.
• CAP and PACELC — the two framings for what a distributed store does during, and outside, a network partition. Settled in Cap stated for this store.

1. Framing: what decision, and what breaks

A key-value store makes exactly one interesting 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. Amazon’s 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 chapter starts. Placing keys is settled by consistent hashing (ch 05). Surviving one dead machine is settled by keeping three copies. Replicas disagreeing, on the other hand, happens continuously, and it is invisible.

The thing that breaks is not availability, it is silence. A store that returns the wrong value quickly and confidently is worse than one that returns an error, because an error wakes somebody up tonight and a stale read becomes a support ticket six weeks later that nobody can trace. So state the design goal as which disagreements the system detects, not as an uptime percentage.

And define the phrase this chapter is about, because it gets used as a shrug. 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.

Read that definition twice, and notice the three things it does not say. It fixes no bound on how long the disagreement lasts. It fixes no bound on how stale a read can be. And it says nothing at all 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.

This chapter is deliberately unsatisfied with the phrase and replaces it, everywhere it can, with a count: 26,000 exposed writes a day (Hinted handoff and sloppy quorum), 900 sibling keys per 60-second partition (Cap stated for this store), a 3-hour hint window, a weekly repair. “Eventually consistent” is a true statement about this store and a useless one; every number that replaces it is the actual answer.

Requirements

Here is what the store must do, and the four targets that later sections are obliged to hit.

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 bug waiting for it.
• put(key, value, context) — the context is not optional, and Vector clocks and the sibling nobody wants is why. delete(key, context) writes a tombstone rather than erasing the row (Lsm trees vs b trees).
• W and R are tunable per request. W is how many of the three copies must acknowledge before a write is called done, and 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 rather than a cluster setting.
• No range scans, no secondary indexes, no transactions spanning more than one key. A range scan is “give me every key between A and M”; a secondary index is a lookup by something other than the key. Say this exclusion out loud early, because giving it up is what pays for everything else.

Non-functional

p99 below means the 99th percentile: the latency that 99 of every 100 requests come in under, so one request in a hundred is slower than the stated number.

Back-of-envelope

Before any mechanism, size the system, because almost every argument later in the chapter is settled by one of these figures.

One notation collision is worth getting ahead of, because both letters are called N in different papers. The Dynamo literature calls the replica count N, while ch 05 calls the fleet size N. In this chapter the fleet is S = 16 and the replica set is N = 3.

Which numbers are borrowed and which are assumed here

This matters because an interviewer will ask where a figure came from, and “I made it up” and “it is the standing setup” are different answers.

Ch 05 fixes S = 16 servers, RF 3, 10.88 TB of logical data and 100,000 writes per second. This chapter reuses those unchanged.

Ch 05 does not state a read rate. The 100,000 reads per second is this chapter’s own addition, and it is not decorative — it sizes the entire bloom-filter argument in The write path the read path and the bloom filter. So label it an assumption rather than inheriting it as a fact.

The two premises underneath the 10.88 TB are also this chapter’s, and they are stated rather than derived anywhere: 500 million users, 20 stored objects each. That product is what the first line of the block below computes. Ch 05 currently describes that line as “derived line by line” here; it is a premise, not a derivation, and a premise is the honest thing to call it — 500 million users is a large consumer product, and 20 objects each is a shopping cart, a profile and some session state. Change either one and every figure in this chapter moves with it.

The arithmetic

The first block sizes storage; the second sizes the network. Nothing here is rounded before it is used, so you can follow any later figure back to a line in this block.

users:                          500,000,000                                  users     (premise)
objects per user:                        20                                  obj/user  (premise)
objects:          500,000,000 x 20                        =  10,000,000,000  objects
record bytes:     32 + 1,024 + 8 + 24                     =           1,088  B/record
logical corpus:   10,000,000,000 x 1,088                  =  10,880,000,000,000  B  = 10.88 TB
replicated RF 3:  10,880,000,000,000 x 3                  =  32,640,000,000,000  B  = 32.64 TB
per node:         32,640,000,000,000 / 16                 =   2,040,000,000,000  B  =  2.04 TB
keys per node:    10,000,000,000 x 3 / 16                 =   1,875,000,000  keys

replica writes/s: 100,000 x 3                             =         300,000  writes/s, whole fleet
per node:         300,000 / 16                            =          18,750  writes/s, one node
per node bytes/s: 18,750 x 1,088                          =      20,400,000  B/s     = 20.4 MB/s
fraction of NIC:  20,400,000 / 125,000,000                =           0.163           = 16.3%

Two lines are worth pausing on. replica writes/s multiplies by 3 because every client write becomes three physical writes — one per replica — so the disks and network cards see three times the client-facing rate. And the last line divides by 125,000,000 because a NIC, the network interface card the machine talks to the network through, is 1 Gbps here, and 1 Gbps is 125 MB/s. That constant is used throughout this chapter.

Read the block as four findings:

1. The logical corpus is 10.88 TB before replication, 32.64 TB after.
2. Each node holds 2.04 TB.
3. Each node holds 1.875 billion keys — the number that sizes bloom-filter RAM in The write path the read path and the bloom filter.
4. The steady write path alone consumes 16% of the network card.

That fourth one is the finding worth saying out loud. The other 84% of the card is the entire budget for three things: repair traffic, replaying writes that were parked elsewhere while a node was down, and moving data when the fleet changes size. All three fire at the same moment during an incident. That is why Bottlenecks and scaling is a section about bandwidth rather than about CPU.

API sketch

The interface is three methods. Read the type annotations rather than the method names — the return type of get and the required context argument on put are arguments the rest of the chapter defends.

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 here are load-bearing. get returns a list, which forces every caller to have an opinion about conflicting values at the moment they write the code, rather than discover the problem in production. And 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,” rather than an accident.

Data model

The record layout is short, and every field in it is spent by a mechanism later in the chapter.

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

Here is where the 1,088 bytes per record used in every estimate above comes from. The 1,024 B value is the payload; the other 64 B is overhead the store adds.

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

The last line is the one to watch. 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, which is why that row is 24 and not more. Vector clocks and the sibling nobody wants is about what happens when it is not the common case, and prices what the extra entries cost.

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 of machines that stores 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. That property is what ch 05 exists to establish.

Tracing one request through the diagram

The diagram below has two halves: the top half is the fleet-level path a request takes, and the boxed subgraph at the bottom is what happens inside one replica once the request lands there.

The client reaches any coordinator node in the fleet. That coordinator computes the preference list and gets three distinct machines — replica s3, replica s7 and replica s11, three arbitrary names — and sends the request to all three.

It then evaluates the one condition the whole design turns on, drawn as the diamond W acks? R responses?. Has a write collected W acknowledgements, or has a read collected R answers? If so, it returns ack / siblings + context: a bare acknowledgement for a write, and 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 instead, and that machine stores a hint (Hinted handoff and sloppy quorum).

Running underneath all of that, gossip + phi-accrual keeps the coordinator’s idea of the membership set current: nodes exchange their view of who is alive, scored by a continuous suspicion level rather than a yes-or-no timeout (Membership gossip and phi accrual).

What the boxed subgraph shows

The subgraph labelled inside one replica is the storage engine, and it runs in four numbered steps (The write path the read path and the bloom filter):

1. WAL append and fsync. fsync is the system call that forces the operating system to push its buffered bytes onto the physical device, so the write survives a power cut.
2. Insert into the memtable, a sorted table held in RAM.
3. Flush the memtable, once full, into an immutable 64 MB SSTable. An SSTable is a sorted string table: a file of key-value pairs written in key order and never modified afterwards.
4. Compaction, the background job that merges those files back together.

Two side structures hang off the engine, drawn with dotted arrows because neither is on the write path:

• A bloom filter per SSTable, which answers “is this key definitely absent from this file?” out of RAM.
• A Merkle tree per key range, a tree of hashes used for anti-entropy — repairing a replica by comparing it with its peers (Merkle trees one bad key in a million). Anti-entropy is the standard name for that background repair: entropy here means the divergence between copies, and this is the process that removes it.

flowchart TB
    C(["client"]) --> CO["coordinator · any node"]
    CO --> PL["preference list<br/>ch 05 · 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

    style M fill:#1d3557,color:#fff
    style S fill:#1d3557,color:#fff
    style B fill:#2d6a4f,color:#fff
    style MT fill:#2d6a4f,color:#fff
    style Q fill:#bc6c25,color:#fff
    style HH fill:#bc6c25,color:#fff

The colours follow the four-colour scheme ch 01 publishes for these same hex values.

Blue is where the authoritative bytes actually live — the memtable and the SSTables, and nothing else. It is emphatically not the Merkle tree. A Merkle tree is a tree of hashes whose entire purpose is to answer questions about the data without touching it; it holds no authoritative copy of anything, and losing one costs you a rebuild, not a byte.

Green is the two side structures that exist to keep work off the request path: the bloom filter, which turns eight disk reads into 0.066, and the Merkle tree, which turns 1.14 GB of streaming into 1,312 bytes.

Orange is the two places the design degrades on purpose — the quorum condition, which returns before every replica has answered, and the sloppy-quorum fallback.

Nothing here is red, because ch 01 reserves red for the one rung you cannot undo, and nothing in this diagram qualifies. A hint is replayed and then deleted, which is about as undoable as a mechanism gets.

2. Deep dive

Seven mechanisms, in the order a request meets them:

1. The quorum — how many copies a call touches (The quorum what w and r actually buy).
2. Vector clocks — the version metadata that detects disagreement (Vector clocks and the sibling nobody wants).
3. Merkle trees — the hash tree that finds disagreement cheaply (Merkle trees one bad key in a million).
4. The storage engine — what actually serves the bytes (The write path the read path and the bloom filter).
5. Hinted handoff — the fallback that keeps writes flowing when a replica is down (Hinted handoff and sloppy quorum).
6. Phi-accrual failure detection — how the cluster decides a replica is down (Membership gossip and phi accrual).
7. CAP — what all of that adds up to during a network partition (Cap stated for this store).

1. The quorum: what W and R actually buy

The W + R > N rule can be derived in one line, and every choice of W and R priced in latency and availability. What matters most, though, is what the rule assumes — because the assumptions are where real systems break.

The rule, and the one-line proof

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 of the N replicas must acknowledge a write and R is how many must answer a read.

The reason is the pigeonhole principle: if you put more items into a set than the set has room for, two of them must land in the same place. Applied here it takes one line. The write landed on some W of the N replicas and the read consults some R of the N. Two subsets of an N-element set must share at least W + R - N members, and W + R > N makes that at least 1.

Substitute the numbers used throughout this chapter. At N = 3 with W = R = 2, the guaranteed overlap is 2 + 2 - 3 = 1 node. At least one replica in the read set is guaranteed to hold the write, so the read cannot miss it.

One convention that flips the intuition

Getting this wrong makes every number below look absurd. The coordinator sends the request to all N replicas and returns as soon as the first W (or R) of them answer.

So R = 1 means the fastest of three, not one node picked at random. Waiting for fewer replicas is a way of discarding slow ones. That is why the arithmetic below runs the opposite direction to most people’s intuition: a smaller W is faster not because it does less work, but because it throws away the stragglers.

What W costs in latency

Suppose one replica takes longer than 10 ms with probability p = 0.01 — a realistic 99th-percentile figure for one node doing a memtable insert plus a WAL fsync. Tail latency is the name for exactly this: the behaviour of the slowest small fraction of requests, which is what users actually notice.

Now the counting argument. The w-th response to arrive is slow precisely when fewer than w of the three were fast, which is the same as saying at least 3 - w + 1 of them were slow. So each line below is a sum of “exactly i replicas were slow” terms, for every i from 3 - w + 1 up to 3.

W = 1:   0.01 x 0.01 x 0.01                              =  0.000001
W = 2:   3 x 0.01 x 0.01 x 0.99 + 0.000001               =  0.000298
W = 3:   1 - 0.99 x 0.99 x 0.99                          =  0.029701

Read each line as a sentence:

• W = 1 — the fastest of three is slow only if all three were slow: 0.01 ^ 3 = 0.000001, one in a million.
• W = 2 — the second of three is slow if exactly two were slow, or all three were. The 3 is the number of ways to choose which two of the three were the slow ones. That is 3 x 0.0001 x 0.99 = 0.000297, plus the 0.000001 from the all-three case, giving 0.000298.
• W = 3 — the last of three is slow if any one of them was slow, which is easier to count as one minus the probability that all three were fast: 1 - 0.99 ^ 3 = 0.029701.

Read that column twice. Waiting for 2 of 3 gives a 10 ms tail of 0.0298%, which is 0.01 / 0.000298 = 34 times better than a single replica’s 1%, because you get to discard the slowest of the three. Waiting for all 3 gives 2.97%, which is 0.029701 / 0.01 = 3 times worse than a single replica.

So a quorum is not a tax on latency; only a full quorum is. Waiting for 2 of 3 is a free hedged request — the technique of issuing redundant copies of an operation and taking the first answer back.

What W costs in availability

Availability moves the same way, only harder. Assume each node is up 99.9% of the time. That is 0.001 x 31,536,000 = 31,536 seconds of unavailability a year, which is roughly 8.8 hours — deploys, restarts, and garbage-collection pauses, not disasters.

A garbage-collection pause is a stall while a managed runtime reclaims memory, during which the process answers nothing. It is the single most common reason a healthy machine looks dead for a few hundred milliseconds.

A write is blocked when fewer than W replicas are up, so it cannot collect its acknowledgements at all. Each line below counts that:

W = 3 blocked:  1 - 0.999 x 0.999 x 0.999                          =  0.002997
W = 2 blocked:  1 - 0.999 x 0.999 x 0.999 - 3 x 0.999 x 0.999 x 0.001  =  0.000003
ratio:          0.002997 / 0.000002998                             =  1,000

Line by line:

• W = 3 blocked — any single node being down blocks the write, so this is one minus the probability all three are up: 1 - 0.999 ^ 3 = 0.002997, about 0.3%.
• W = 2 blocked — the write survives all three being up and exactly two being up, so subtract both from 1. The second term 3 x 0.999 x 0.999 x 0.001 = 0.002994 is “exactly two up,” with the 3 again counting which node is the down one. What is left is 0.000002998, about 0.0003%.
• ratio — 0.002997 / 0.000002998 = 1,000.

W = 3 is a thousand times less available than W = 2, and buys nothing that W = 2, R = 2 does not already give. It is the single most common wrong answer in this interview.

The full menu at N = 3

Five configurations, with the tail and availability arithmetic above applied to each. The W+R>3 column says whether the overlap guarantee holds. “Write blocked” is the probability from the previous block — 3e-6 means 0.000003. Bold rows are the ones worth defending in an interview.

Read (2,2) against (3,1) in both directions, because a single “100x better” is half the sentence. On writes (2,2) is 2.9701 / 0.0298 = 99.7, so call it 100x better. On reads it is 0.0298 / 0.0001 = 298, so it is 298x worse. Neither configuration 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, and (2,2) splits it evenly across both. That trade is the point of the table, and quoting only the favourable half of it is the mistake the table exists to prevent.

W = N, R = 1 is the read-optimized store because the read has nothing left to check — every replica already has every write, so the first answer back is correct. W = 1, R = N is its mirror: the write escapes after one ack and the read pays to reassemble. You are not choosing between fast and slow; you are choosing which of the two paths eats the third replica’s tail latency.

The two functions below are the whole table as code, so you can re-derive any cell rather than memorising it. comb(n, i) is the binomial coefficient — the number of ways to choose i items from n, which is where the 3 in 3 x 0.01 x 0.01 x 0.99 came from. tail_beyond sums the “at least n - w + 1 replicas were slow” terms; write_available sums the “at least w replicas are up” terms. The assertions reproduce every figure quoted above.

from math import comb


def tail_beyond(p, n, w):
    """P(the w-th of n responses is slow), each replica slow w.p. p. The w-th
    is slow exactly when fewer than w were fast: at least n - w + 1 were slow."""
    return sum(comb(n, i) * p ** i * (1 - p) ** (n - i)
               for i in range(n - w + 1, n + 1))


def write_available(a, n, w):
    """P(at least w of n replicas are up), each up independently w.p. a."""
    return sum(comb(n, i) * a ** i * (1 - a) ** (n - i) for i in range(w, n + 1))


assert abs(tail_beyond(0.01, 3, 1) - 0.000001) < 1e-9
assert abs(tail_beyond(0.01, 3, 2) - 0.000298) < 1e-9
assert abs(tail_beyond(0.01, 3, 3) - 0.029701) < 1e-9
assert abs((1 - write_available(0.999, 3, 3)) - 0.002997) < 1e-8
assert abs((1 - write_available(0.999, 3, 2)) - 0.000003) < 1e-8

The four assumptions under all of it

Every number above rests on four assumptions, and each one fails in a specific, nameable way. This is the part interviewers probe, and the part that decides whether the arithmetic survives contact with a real cluster.

Assumption 1: replicas are slow independently. Both tail_beyond and write_available multiply per-replica probabilities together, which 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 touches replicas in sequence. A shared top-of-rack switch drops packets for all of them. A compaction storm triggered by the same traffic pattern hits all three at once. A coordinated garbage-collection pause across a homogeneous fleet is common.

When failures correlate, 1 - 0.999^3 = 0.002997 collapses toward the single-node figure of 0.001, and the advertised thousand-fold advantage of W = 2 over W = 3 shrinks toward nothing. The mitigation is placement — put the three replicas in three racks or three availability zones — which is why rack-aware placement appears in the requirements table rather than as an afterthought.

Assumption 2: p = 0.01 and per-node availability of 99.9% are stable inputs. They are measurements, not constants, and they move with load. A node at 90% disk utilization has a different p than the same node at 40%, so treat the table as a shape — full quorum is worse than partial quorum, by roughly two orders of magnitude — rather than as five decimal places you can quote in a capacity plan.

Assumption 3: the read set and the write set are drawn from the same N replicas. This is the load-bearing one, because the pigeonhole argument is a statement about two subsets of one set. The moment a write is allowed to land on a machine outside the preference list, the two sets are drawn from different pools and can be disjoint no matter what W + R sums to. That is exactly what sloppy quorum does, and Hinted handoff and sloppy quorum quantifies the resulting exposure at about 26,000 writes a day.

Assumption 4: “the last completed write” is a well-defined thing. It is not, when two clients write at the same instant. The overlap guarantee tells you the read set contains a replica that acked the last completed write; it says nothing about which of two concurrent writes should win, which is why Vector clocks and the sibling nobody wants exists.

What W = 1, R = 1 actually gives up

Not “consistency” as an abstraction. Something countable.

The write is acknowledged as soon as the fastest replica has it, while the other two copies are still travelling. The interval between the acknowledgement and the last replica receiving the write is the replication window.

A read issued inside that window lands on the one replica that already has the value with probability 1 / 3 — and that 1/3 is a lower bound, not the exact figure. It is the number for a replica picked uniformly at random. But the convention set above is that R = 1 returns the fastest of three, and the replica that acknowledged the write first is, for the same reasons of load and proximity, disproportionately likely to answer the read first too.

So real freshness sits somewhere between 1/3 and 1, and 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. Quote it as a bound, or state that you are using the random-replica model. Do not present 1/3 as a measurement.

The window is normally under a millisecond. But it has no upper bound during a garbage-collection pause, a network partition, or a slow disk, and nothing in the protocol notices or reports it. (1,1) is therefore not “slightly weaker.” It is “correct except when something is wrong,” which is the definition of a bug you only meet during an outage.

Quorum does not give you linearizability

Linearizability is the strongest single-object guarantee there is: every operation appears to take effect at one instant between its call and its return, so once a read returns a new value, no later read ever returns an older one.

Quorum overlap gives you something much narrower — “the read set intersects the set of replicas that acknowledged the last completed write.” Four gaps remain between that and linearizability:

1. A write still in flight is neither completed nor absent, so two successive reads can see old, then new, then old again.
2. A write that reached one replica and then failed is never rolled back, and may surface later as a value nobody thinks they wrote.
3. Two writes issued at the same time still need Vector clocks and the sibling nobody wants to order them.
4. Sloppy quorum (Hinted handoff and sloppy quorum) breaks the intersection argument outright.

Two weaker guarantees are achievable here, and are worth naming precisely.

Read-your-writes means a client that just wrote a value will see that value on its own next read. You get it if that client’s reads and writes go through one coordinator.

Monotonic reads means a client never sees time run backwards: having seen a new value, it will not later see an older one. You get that only once you add read repair — a term this chapter leans on four more times, so define it here. Read repair is the coordinator noticing, while it is already holding the replicas’ answers to a read, that some of them are behind, and writing the up-to-date value back to those replicas on the spot. It costs no extra requests, because the responses were already paid for by the read itself. Hinted handoff and sloppy quorum prices it and states what it cannot see.

So the honest sentence is: “quorum gives me read-your-writes if I route through one coordinator, and monotonic reads only if I add read repair. It does not give me a linearizable register.”

A linearizable register needs consensus — a protocol such as Raft or Paxos in which a majority of nodes agree on a single ordered log of operations before any of them is applied (Alternatives rejected). The related question of what guarantees a transaction gets is Transactions acid precisely.

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 one wins; that detection has a per-record cost, and the standard trick for bounding the cost can silently lose data.

The definition, in four sentences

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 is a compact record of “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 that appears in B. In plain terms: whoever produced A had already seen everything B contains.

If neither clock descends from the other, then neither writer had seen the other’s write. The two writes are concurrent, and both must be kept. Two values kept side by side like that are siblings.

Note carefully what the clock does and does not do. It detects concurrency exactly, and it never decides a winner.

A trace of two concurrent writes

The diagram below is the whole mechanism in eight messages. The key is cart:42. Two clients write it through two different coordinators, Sx and Sy. Both machines are healthy and reachable, and there is no network partition anywhere — concurrency needs nothing to go wrong.

Watch the clock values in the reply arrows, and read the three Note boxes: the first is where the store decides the writes are concurrent, the last is where the conflict disappears.

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

Walk the last two steps slowly, because they are the mechanism.

The merged context [(Sx,1),(Sy,1)] is the pointwise maximum of the siblings’ clocks: take each node name in turn and keep the larger of the two counters. Sx appears as 1 in one sibling and is absent (so, 0) in the other, giving 1. Same for Sy.

When the client writes back with that context, Sx increments its own entry, and the result is [(Sx,2),(Sy,1)]. Check it against the descends rule: it has Sx = 2 >= 1 and Sy = 1 >= 0 versus the first sibling, and Sx = 2 >= 0 and Sy = 1 >= 1 versus the second. It descends from both, 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 said what the merge should be. That is the whole design: the database’s job is to lose neither write and to refuse to guess between them.

The mechanism as code

The listing below is the store’s entire conflict machinery. Four small functions do the detection — descends, concurrent, merge_clocks and coalesce — and quorum_read is what a coordinator actually runs when the replicas answer.

Read the docstrings on coalesce and quorum_read before the code. They are not commentary; each one names a specific way this function is written wrong in real systems, and the assertions at the bottom of the block are the cases that catch it. Four of those decisions are unpacked in prose right after the listing.

from dataclasses import dataclass, field


def descends(a, b):
    """a has seen everything b has seen."""
    return all(a.get(n, 0) >= c for n, c in b.items())


def concurrent(a, b):
    return not descends(a, b) and not descends(b, a)


def merge_clocks(clocks):
    out = {}
    for c in clocks:
        for n, v in c.items():
            out[n] = max(out.get(n, 0), v)
    return out


@dataclass
class Version:
    value: bytes
    clock: dict = field(default_factory=dict)


def coalesce(versions):
    """Drop every version some other version strictly descends from.

    Deduplicate on (clock, value), never on the clock alone. Two writes can
    carry the SAME clock and different values -- that is exactly what a
    blind write from one client produces, and it is the case dotted version
    vectors exist to fix. Keying dedup on the clock discards one of them
    silently, which is the failure this section recommends DVVs against.

    An EMPTY clock is the second trap and it is the API's own promise.
    `descends(anything, {})` is True for every clock, because "has seen
    everything the empty set contains" is vacuously true -- so an empty
    context is an ancestor of every version in the store, and the plain
    rule would drop it. That is precisely the write the API section says
    cannot be lost: passing an empty context is a declaration of "I accept
    that my write will become a sibling", not a declaration that it may be
    discarded. A real coordinator never lets the case arise, because it
    stamps its own (node, counter) onto the write before storing it, so an
    empty clock is only ever seen in flight. This function does not get to
    assume that, so it refuses to treat an unstamped write as an ancestor.
    """
    keep = []
    for v in versions:
        if v.clock and any(descends(o.clock, v.clock) and o.clock != v.clock
                           for o in versions):
            continue
        if any(v.clock == k.clock and v.value == k.value for k in keep):
            continue
        keep.append(v)
    return keep


def quorum_read(responses, r):
    """responses: [(node, [Version, ...])] in arrival order. Returns the
    siblings, the context to write back with, and the replicas to repair.

    The quorum is the first R responses -- that is what the read waits for.
    The repair list is every replica that ANSWERED and is missing a sibling,
    including the late ones and including a replica that holds nothing at
    all. Read repair is free-riding on a response you already have; scoping
    it to the quorum leaves known-stale replicas known-stale.
    """
    if len(responses) < r:
        raise TimeoutError(f"only {len(responses)} of R = {r} replicas answered")
    quorum = responses[:r]
    siblings = coalesce([v for _, vs in quorum for v in vs])
    context = merge_clocks([s.clock for s in siblings])
    repair = [n for n, vs in responses
              if not all(any(descends(v.clock, s.clock) for v in vs) for s in siblings)]
    return siblings, context, repair


d1 = Version(b"{milk}", {"Sx": 1})
d2 = Version(b"{eggs}", {"Sy": 1})
assert concurrent(d1.clock, d2.clock)

# R = 2: the first two responses disagree, so BOTH come back to the client.
sibs, ctx, repair = quorum_read([("s1", [d1]), ("s2", [d2]), ("s3", [d1])], r=2)
assert [s.value for s in sibs] == [b"{milk}", b"{eggs}"]
assert ctx == {"Sx": 1, "Sy": 1}
# s1 and s2 each hold one sibling; s3 answered late holding only {milk}, so
# it is stale too and read repair must reach it.
assert repair == ["s1", "s2", "s3"]

# The replica that holds NOTHING is the one scoped repair misses: it is never
# in the first R responses when it is the slow one, and it is the most stale.
sibs, ctx, repair = quorum_read([("s1", [d1]), ("s2", [d2]), ("s3", [])], r=2)
assert repair == ["s1", "s2", "s3"]
assert ctx == {"Sx": 1, "Sy": 1}

# A replica that already holds a descendant of every sibling needs nothing.
merged = Version(b"{milk,eggs}", {"Sx": 2, "Sy": 1})
sibs, ctx, repair = quorum_read([("s1", [d1]), ("s2", [d2]), ("s3", [merged])], r=2)
assert repair == ["s1", "s2"]

# --- identical clocks, different values: two siblings, not one --------
# Both clients read the same context and wrote blind through the same
# coordinator, so the two clocks are EQUAL. `descends` is >=, so each one
# descends from the other, in both directions -- which is exactly why
# `concurrent` is False on the next line. The blind spot is not that the
# clock test calls them ordered; it is that the clock test cannot tell the
# two values apart AT ALL, in either direction, so it can say nothing
# useful about them and the values themselves are the only evidence left.
e1 = Version(b"{milk}", {"Sx": 1})
e2 = Version(b"{eggs}", {"Sx": 1})
assert descends(e1.clock, e2.clock) and descends(e2.clock, e1.clock)
assert not concurrent(e1.clock, e2.clock)   # ...which is why this is False
assert coalesce([e1, e2]) == [e1, e2]       # ...so the VALUES must both survive
assert coalesce([e1, e1]) == [e1]           # a genuine duplicate still collapses

# --- the empty context, which is a declared blind write and not garbage ---
# descends(anything, {}) is True, so the unguarded rule makes an unstamped
# write an ancestor of everything and drops it -- the exact loss the API
# section promises an empty context does NOT cause.
blind = Version(b"{bread}", {})
assert descends(d1.clock, blind.clock)      # every clock descends the empty one
assert coalesce([d1, blind]) == [d1, blind]     # both survive
stamped = Version(b"{bread}", {"Sz": 1})    # what a coordinator actually stores
assert coalesce([d1, stamped]) == [d1, stamped]

sibs, ctx, repair = quorum_read([("s1", [e1]), ("s2", [e2])], r=2)
assert [s.value for s in sibs] == [b"{milk}", b"{eggs}"]
# ...and read repair is blind here: every clock descends every other clock,
# so nothing looks stale even though each replica is missing a value.
assert repair == []

# The client reconciles and writes with that context. Both siblings collapse.
d3 = Version(b"{milk,eggs}", {**ctx, "Sx": ctx["Sx"] + 1})
assert d3.clock == {"Sx": 2}
assert descends(d3.clock, e1.clock) and descends(d3.clock, e2.clock)
assert coalesce([e1, e2, d3]) == [d3]

Four properties of that code carry the argument, and each one is a decision that is easy to get backwards.

1. coalesce deduplicates on the pair (clock, value), not on the clock alone. Two writes can carry the identical clock and different values. That is what happens on a blind write — a put that was not preceded by a get, so the client had no context to build on and both clients started from the same one.

When the clocks are equal, no clock comparison can distinguish the two values. descends reports that every replica already has everything, and read repair therefore finds nothing to fix. In that state the store’s only defence is to keep both values, so deduplicating on the clock alone would discard a real write with no error and no trace. This is precisely the case dotted version vectors, recommendation 2 below, exist to handle.

2. quorum_read repairs every replica that answered, not just the first R. The read waits for R responses — that is what R means. But a response arriving after the read has already returned is free information, and a replica that holds nothing at all for the key is simultaneously the most stale and the least likely to be among the first R to answer.

Scoping repair to the quorum would leave known-stale replicas known-stale, which contradicts the claim in Hinted handoff and sloppy quorum that read repair fixes exactly the keys people are reading.

3. The repair list is empty when it ought to be. A replica that already holds a descendant of every sibling is not written back to. So in the overwhelmingly common case — all three replicas agreeing — the mechanism costs nothing at all.

4. An empty clock is never treated as an ancestor. descends(anything, {}) is True for every clock, because “has seen everything the empty set contains” is vacuously true.

Left unguarded, that makes an empty-context write an ancestor of every other version, and coalesce drops it silently. That is the exact loss the API sketch promises an empty context does not cause: passing one is an explicit statement of “I accept that my write will become a sibling,” not consent to have it discarded.

A production coordinator never lets the case reach this function, because it stamps its own (node, counter) onto the write before storing it. But then the guarantee lives in the coordinator, not in the merge rule — and this listing has no coordinator. State which of the two is carrying the guarantee in your design, because exactly one of them has to.

The cost nobody volunteers: 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. But under coordinator failover, sloppy quorum, or a client that reconnects to a different node each time, it becomes however many nodes the key has ever touched. At S = 16 the ceiling is 16 entries; in a 1,000-node cluster it is 1,000.

Price a 10-entry clock against the 1,088-byte record from the data model. The baseline record already carries one entry, so the growth is nine extra entries, not ten:

clock entry:      8 + 8 + 8                              =  24    B   (node id, counter, timestamp)
10-entry clock:   10 x 24                                =  240   B
record grows by:  240 - 24                               =  216   B   (the 9 entries it did not have)
                  216 / 1,088                            =  0.199       = +19.9%

A 10-entry clock is +19.9% on every record, and its 240 bytes are 23% of the 1,024-byte payload it describes (240 / 1,024 = 0.234). At 1.875 billion keys per node that is real disk, real network on every read, and real CPU on every comparison.

Why the standard fix can lose data

The standard fix is Dynamo’s: cap the clock at 10 entries and evict the entry with the oldest timestamp. It is lossy, and not only in the harmless direction.

The harmless case is a descendant that loses an entry its own ancestor still has. The two then look concurrent, and you get a spurious sibling — resolvable, safe, nothing lost.

The other case is not. In the block below, two genuinely concurrent writes go in and one comes out an ancestor of the other. Read the four assertions in order; the third one is the damage.

a = {"S1": 1, "S2": 1}          # client A's write, via S1 then S2
b = {"S1": 1, "S3": 1}          # client B's write, via S1 then S3
assert concurrent(a, b)         # genuinely concurrent: both must survive

a_trunc = {"S1": 1}             # truncation evicted (S2, 1) as the oldest entry
assert descends(b, a_trunc)     # b now DOMINATES a
assert not descends(a_trunc, b)
# coalesce() will now discard a as an ancestor of b. A's write is gone,
# with no sibling, no error, and no way to detect it after the fact.

Truncation can turn two concurrent writes into a false ancestor relationship and silently drop one of them. The Dynamo paper is honest that truncation harms reconciliation accuracy; the sharp version is that it can lose data.

It almost never fires — Dynamo reported that 99.94% of requests over 24 hours saw exactly one version — but “almost never” plus “undetectable” is the profile of the bugs that take a year to find.

Three ways out, in the order to offer them in an interview:

1. Keep the clock server-side and per-replica, not per-coordinator. The entry count is then bounded by N = 3 rather than by the fleet. This is the version vector, and it is what makes the growth problem mostly theoretical in a well-routed system.
2. Dotted version vectors, usually shortened to 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) pair identifying that one specific write — next to the summarising vector, which lets the store tell “this is a fresh write from a client that had seen version 5” apart from “this is a genuinely concurrent write,” so repeated blind writes from one client stop manufacturing siblings.
3. Give up on detection and use last-write-wins, abbreviated LWW, which is Cassandra’s default. There is no metadata and no client-side reconciliation logic at all. 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, the Network Time Protocol, which typically holds machines within about 10 ms of each other inside a datacenter — so two writes 2 ms apart can be ordered backwards, and the loser leaves no trace. LWW is correct for data that is immutable, or idempotent in the sense that applying the same write twice leaves the same result. It is wrong for a shopping cart, a counter, or anything a human will later ask “where did my edit go?” about.

What a vector clock assumes, and what fails when it does not hold

It assumes writes are routed through a bounded set of coordinators. The entry count is exactly the number of distinct coordinators that have touched the key, so when routing is unstable the metadata cost grows toward the fleet size and the +19.9% figure above becomes far worse.

It assumes the client can merge. The store’s entire strategy is to hand conflicts back. If the application has no merge rule, siblings simply accumulate until someone writes siblings[0] and loses data — which is the trigger for moving to consensus, in Alternatives rejected.

It assumes truncation never fires. When it does, two genuinely concurrent writes can be turned into a false ancestor-descendant pair, and one of them vanishes with no error.

And LWW assumes clocks are ordered correctly across machines, which NTP does not guarantee at millisecond resolution. When that fails, the later write loses and nothing anywhere records that it happened.

3. Merkle trees: one bad key in a million

Two machines holding a million keys each can find the one key they disagree about while exchanging about a kilobyte. Three costs make the technique a scheduled job rather than a continuous one.

The question, and the naive answer

Two replicas hold the same range of keys. Are they identical?

The obvious answer is to stream the whole range across and compare it. That works and is unaffordable.

The efficient answer is a Merkle tree, also called a hash tree. It is a binary tree built like this: 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 whole point: the single node at the top, the root, is a 32-byte fingerprint of the entire range.

Take a range of 2 ^ 20 = 1,048,576 keys and cost the naive approaches first. The first two lines below are streaming; the last three describe the tree you would build instead.

streaming the range:  1,048,576 x 1,088                  =  1,140,850,688  B  = 1.14 GB
at 125 MB/s:          1,140,850,688 / 125,000,000        =           9.13  seconds
one digest per key:   1,048,576 x 32                     =     33,554,432  B  = 33.6 MB
tree depth:           log2(1,048,576)                    =             20  levels below the root
tree nodes:           2 x 1,048,576 - 1                  =      2,097,151  nodes total

Streaming the range costs 1.14 GB and 9.13 seconds of a completely saturated network card — and that is per range, per pair of peers. Shipping only a digest (one 32-byte hash per key instead of the key’s full contents) still costs 33.6 MB.

Both grow in direct proportion to the number of keys, written O(n). That proportionality is why nobody compares replicas either way.

Two structural facts about the tree are worth holding onto. A binary tree over 2^20 leaves is 20 levels deep, because each level halves the count. And a binary tree with L leaves has 2L - 1 nodes in total, so this one has 2,097,151 — a number that comes back when you consider shipping the whole tree.

The comparison, and why it is O(log n)

If the two roots match, 1,048,576 keys are proven identical by exchanging 32 bytes. That is the common case: most range comparisons find nothing wrong, and they cost one hash.

When the roots differ, you descend. At each level you only need to compare the two children of a node you already know is bad — the other subtree is proven clean and never touched. So the total is the root, plus two hashes per level:

hashes compared:      1 + 2 x 20                          =  41     hashes
bytes on the wire:    41 x 32                             =  1,312  B

41 hashes and 1,312 bytes localise one differing key among 1,048,576, against 1,048,576 comparisons and 1.14 GB for the streaming approach.

Cost that grows with the logarithm of the key count rather than with the key count itself is written O(log n). Those two numbers are what that notation means in practice: doubling the range adds two hashes, not a million.

The tree in code

build constructs the tree bottom-up and returns a list of levels, leaves first. diff walks two trees downward together, keeping a frontier of node indices known to be bad and expanding each one into its two children.

The line to watch in diff is nxt.extend([2 * idx, 2 * idx + 1]) — that is the descent, and the fact that it only runs inside the if a[lvl][idx] != b[lvl][idx] branch is the entire saving. The assertions at the bottom corrupt exactly one leaf out of 1,048,576 and check that the walk finds it in 41 comparisons.

import hashlib


def build(leaf_hashes):
    """Bottom-up. levels[0] is the leaves, levels[-1] is [root]."""
    levels = [leaf_hashes]
    while len(levels[-1]) > 1:
        cur = levels[-1]
        levels.append([hashlib.blake2b(cur[i] + cur[i + 1], digest_size=32).digest()
                       for i in range(0, len(cur), 2)])
    return levels


def diff(a, b):
    """Descend both trees together. Returns (differing leaves, hashes compared)."""
    frontier, compared = [0], 0
    for lvl in range(len(a) - 1, -1, -1):
        nxt = []
        for idx in frontier:
            compared += 1
            if a[lvl][idx] != b[lvl][idx]:
                nxt.extend([idx] if lvl == 0 else [2 * idx, 2 * idx + 1])
        frontier = nxt
        if not frontier:
            break
    return frontier, compared


n = 1 << 20
leaves = [hashlib.blake2b(f"{i}:v1".encode(), digest_size=32).digest() for i in range(n)]
peer = list(leaves)
peer[738_291] = hashlib.blake2b(b"738291:v2", digest_size=32).digest()

ta, tb = build(leaves), build(peer)
assert len(ta) - 1 == 20 and sum(len(x) for x in ta) == 2_097_151
bad, compared = diff(ta, tb)
assert bad == [738_291]
assert compared == 1 + 2 * 20 == 41

Round trips are a different question from hashes

How many network messages this takes is separate from how many hashes it compares, and it is the one interviewers push on.

A round trip is one message out and its reply back. Inside a single datacenter that costs about 0.5 ms (ch 02).

Take the two extremes first. Descending one level per message costs 20 + 1 = 21 messages, so 21 x 0.5 = 10.5 ms of pure waiting — cheap in bytes, expensive in latency. Going the other way and shipping the entire tree in one message costs 2,097,151 x 32 = 67,108,832 bytes, or 67 MB, which for a small range is worse than just sending the data.

The workable answer sits in between: ship the top 11 levels in one message, then the one surviving subtree in a second.

top 11 levels:        2 ^ 11 - 1                          =      2,047  nodes
                      2,047 x 32                          =     65,504  B
two messages:         2 x 65,504                          =    131,008  B  = 131 KB
vs streaming:         1,140,850,688 / 131,008             =      8,708  x less traffic

Why 11 levels is 2,047 nodes: a full binary tree of d levels has 2^d - 1 nodes, and 2^11 - 1 = 2,047.

Why two messages suffice: the deepest level in that first message holds 2^10 = 1,024 nodes, and each of them covers 1,048,576 / 1,024 = 1,024 leaves. Once you know which of those 1,024 nodes mismatches, its subtree is another 11 levels — the same size as the first message — so the second message finishes the job.

Two round trips, 131 KB, 8,708x less traffic than streaming the range.

Now scale that from one range to a whole node. Each node holds 1.875 billion keys, so it holds many ranges of 1,048,576 keys apiece:

ranges per node:      1,875,000,000 / 1,048,576           =          1,788  ranges
roots on the wire:    1,788 x 32                          =         57,216  B  = 57 KB
bytes of data per byte of hash:
                      2,040,000,000,000 / 57,216          =     35,654,362  B of data per B of hash

57 KB of root hashes proves that 2.04 TB agrees with a peer — a 35-million-to-one compression of the question.

The three costs

The technique is not free, and an interviewer expects at least the first of these unprompted.

• Building the tree is O(n) and it is a full range read. 2.04 TB at 1 GB/s sequential is 2,040,000,000,000 / 1,000,000,000 = 2,040 seconds, 2,040 / 60 = 34 minutes per node, 16 x 34 = 544 minutes = 9.1 hours for the fleet. Cheap to compare, expensive to build — which is exactly why repair is a scheduled weekly job and not a continuous process.
• Any write invalidates the path from that leaf to the root. Maintain it incrementally (20 hash updates per write, 18,750 x 20 = 375,000 per node per second — trivial CPU) or rebuild at repair time. Cassandra rebuilds.
• Real implementations cap the depth, and the cap costs you over-streaming. Cassandra caps at 2 ^ 15 = 32,768 leaves, so each leaf covers 1,048,576 / 32,768 = 32 keys and one differing key streams 32 x 1,088 = 34,816 bytes instead of 1,088 — a 32x over-stream, fine. At 100 M keys per range with the same cap a leaf covers 3,052 keys and one bad key streams 3.3 MB, which is why “repair streams far more than my drift” is the classic Cassandra complaint. The fix is more, smaller ranges, not a deeper tree.

What the Merkle comparison assumes, and what fails when it does not hold

It assumes both peers build their trees over identical range boundaries. Leaves must cover the same key intervals on both sides, or the roots differ for a reason that has nothing to do with the data and every comparison finds a phantom mismatch. That is exactly why a topology change that redraws range boundaries invalidates every tree in the cluster, and why Bottlenecks and scaling insists on one topology change at a time.

It assumes the two trees describe the same instant. Live writes arriving mid-comparison change leaves under you, so a repair run reports differences that were never divergence. The standard mitigation is to build both trees from a snapshot.

It assumes hash collisions are impossible in practice. For a 32-byte hash they effectively are, but it is worth naming: two different values with the same leaf hash would be reported as identical, and the divergence would never be repaired.

And it assumes the number of differing keys is small. Every cost above is derived for one bad key. If half the range differs, the descent visits most of the tree, the traffic saving evaporates, and streaming the range would have been cheaper — which is exactly the regime a node that has been down for a day is in.

4. The write path, the read path, and the bloom filter

Away from the fleet and into one replica: what a node actually does with a write and a read, why this traffic forces a specific compaction strategy, and why the bloom filter is not an optimization but the thing that makes the read path possible at all.

The write path

Inside a single replica, a write is three steps on the critical path and two more in the background.

On the critical path: the node appends the change to the write-ahead log and calls fsync so the bytes are on the physical device; it inserts the key into the memtable; and it acknowledges the write. That is why a write costs one sequential disk append rather than a random one — nothing is searched for, nothing is updated in place.

In the background: when the memtable fills, it is flushed to an SSTable, and compaction merges those accumulating SSTables back together, dropping superseded versions and expired tombstones.

The general mechanics, the write-amplification formula, and the tombstone problem are all derived in Lsm trees vs b trees. What is new here is what this traffic does to them. Start with how often that background flush fires, using the 20.4 MB/s per node from the back-of-envelope:

memtable fill:        64,000,000 / 20,400,000             =    3.14   seconds to fill 64 MB
SSTables per day:     86,400 / 3.14                       =  27,516   files per node per day

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

Choosing a compaction strategy, in bandwidth

The two strategies differ in how they arrange those files:

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

Leveled compaction’s cost depends on how many levels you have, so count them. Starting from a 64 MB memtable and growing by a size ratio of T = 10 per level, each level is:

L1:  64,000,000 x 10                                     =            640,000,000  B  = 640 MB
L2:  640,000,000 x 10                                    =          6,400,000,000  B  = 6.4 GB
L3:  6,400,000,000 x 10                                  =         64,000,000,000  B  =  64 GB
L4:  64,000,000,000 x 10                                 =        640,000,000,000  B  = 640 GB
L5:  640,000,000,000 x 10                                =      6,400,000,000,000  B  = 6.4 TB

L5 at 6.4 TB is the first level that can hold this node’s 2.04 TB, so L = 5 levels.

Now apply the leveled write-amplification formula from Lsm trees vs b trees, which is 1 (WAL) + 1 (L0 flush) + T x L — one write for the log, one for the initial flush, and T rewrites per level as data is merged downward:

leveled write amp:    1 + 1 + 10 x 5                      =            52  B written per B stored
disk write bytes/s:   20,400,000 x 52                     = 1,060,800,000  B/s = 1,061 MB/s
size-tiered write amp (sql 03 section 6 table: ~4-5x)          4.5          B written per B stored
disk write bytes/s:   20,400,000 x 4.5                    =    91,800,000  B/s =    92 MB/s

Leveled compaction needs 1,061 MB/s of write bandwidth against a 1 GB/s NVMe device, so it does not fit at all — before a single read has been served. (An NVMe is a solid-state drive attached directly to the PCIe bus, the fast end of what a server has.) Size-tiered needs 92 MB/s, which is 9% of the device.

That is the entire reason 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 that choice twice. Once on the read path, which is the next subsection. And once in space amplification — how many bytes of disk are occupied per byte of live data. Size-tiered runs at about 2x, so provision 2 x 2.04 = 4.08 TB of disk per node.

The read path, and why it needs a filter

The read path is the mirror image of the write path. Check the memtable first, then each SSTable, newest file first.

Because size-tiered compaction leaves several overlapping SSTables, call that number 8. Without any filter, that is 8 random disk reads for every lookup — and most of them find nothing at all, because a given key lives in at most one SSTable until compaction merges them. Seven of every eight reads are guaranteed misses.

A bloom filter is the fix. It is a small array of bits in RAM, one per SSTable, that answers exactly one question: “is this key definitely not in this file?” It can answer “definitely not present” or “maybe present,” never “definitely present.”

The mechanism is two lines. You add a key by hashing it to k positions in the array and setting those bits. You query a key by hashing it the same way and checking whether all k bits are set.

Because bits get shared between keys, a key that was never added can find all k of its bits already set by other keys. That is a false positive: the filter says “maybe,” and the disk read it triggered turns out to be wasted. It never produces a false negative, because a key that was added always has its bits set.

Sizing the filter

Two symbols run through the rest of this subsection: m is the number of bits in the array, n is the number of keys stored in it, and the design knob is their ratio m/n, the bits per key. k is the number of hash positions probed per key.

The false-positive rate follows in two steps:

P(one bit still 0 after kn insertions)  =  (1 - 1/m)^(kn)  ~=  e^(-kn/m)
P(all k probed bits are 1)              =  (1 - e^(-kn/m))^k

The first line asks: after kn bit-settings, what is the chance a particular bit was missed every time? The second line is the false-positive rate — the chance that all k bits a never-added key probes happen to be set already.

Differentiating the second line with respect to k gives the optimum k = (m/n) ln 2. At that point each bit is 1 with probability exactly 1/2, which makes the filter maximally uninformative per bit — that is what “optimal” means here.

Substitute 10 bits per key, step by step:

optimal k:            10 x 0.693147                       =  6.93        (= (m/n) ln 2)
use k = 7, so kn/m:   7 / 10                              =  0.7         (k is an integer)
1 - e^(-0.7):         1 - 0.496585                        =  0.503415    (chance one probed bit is 1)
FP rate:              0.503415 ^ 7                        =  0.008193    (all 7 are 1)

The false-positive rate is 0.82% at 10 bits per key, which is the “~1%” quoted in Lsm trees vs b trees.

Substituting the optimal k back into the formula collapses it to a closed form, p = 0.6185 ^ (m/n), and that one expression produces the whole curve. In the table below, read down the FP-rate column to see how fast it falls, and across to the last column to see what each step costs in RAM at this chapter’s 1.875 billion keys per node.

Two things make that table memorizable rather than something to look up.

0.6185 ^ 4.8 = 0.0996, so every 4.8 additional bits per key divides the false-positive rate by 10. That also explains why nobody goes past about 16 bits: you pay 4.8 bits per key of RAM per order of magnitude, and by then the disk reads you save are already noise.

The size column is just n x (m/n) / 8 bytes. At 10 bits/key that is 1,875,000,000 x 10 / 8 = 2,343,750,000 bytes, or 2.34 GB, which is 2,343,750,000 / 64,000,000,000 = 0.0366 — under 4% of a 64 GB box.

The payoff, in disk reads and then in devices

Price the worst common case: a key absent from all 8 SSTables. That is the “does this user exist” lookup, and it is also what happens at every SSTable that does not hold a key that does exist.

no filter:                                                   8        disk reads
with filter:          8 x 0.008193                        =  0.0655   disk reads
P(any false positive): 1 - 0.991807 ^ 8                   =  0.0637   of lookups touch disk
key present in the oldest of 8:  1 + 7 x 0.008193         =  1.0574   disk reads

Line by line: without a filter every lookup reads all 8 files. With a filter, each of the 8 is read only on a false positive, so the expected count is 8 x 0.008193 = 0.0655. The third line is the chance that at least one of the 8 filters lies, computed as one minus the chance none of them does. The fourth line is the case where the key genuinely exists in the oldest file: one real read plus a false positive on each of the other 7.

8 disk reads become 0.066 — a 8 / 0.0655 = 122x reduction — and 93.6% of absent-key lookups touch no disk at all (1 - 0.0637 = 0.9363).

Now translate reads into hardware, using this repo’s standing 100 MB/s random-read figure and a 4 KB page:

device random reads/s:  100,000,000 / 4,096               =  24,414   IOPS per device
per node read requests: 100,000 x 3 / 16                  =  18,750   lookups/s per node
no filter:              18,750 x 8                        = 150,000   IOPS needed
                        150,000 / 24,414                  =    6.14   devices
with filter, key present: 18,750 x 1.0574                 =  19,826   IOPS needed
                        19,826 / 24,414                   =    0.81   devices
with filter, key absent: 18,750 x 0.0655                  =   1,228   IOPS needed

Without bloom filters this workload needs 6.1 devices per node just to serve reads. With them it needs 0.81 of one device.

That is why calling bloom filters an optimization is the wrong framing: they are what makes the LSM read path viable at all. Six devices per node is not a tuning problem, it is a different machine.

Two properties are worth naming unprompted.

The filter produces false positives but never false negatives, so a “no” is a proof and the SSTable is skipped without touching the disk, while a “maybe” costs at most one wasted read.

And a bloom filter cannot answer range queries. It only knows about exact keys, which is the gap Lsm trees vs b trees identifies as the LSM tree’s structural weakness.

What the bloom arithmetic assumes

State these unprompted, because a real cluster violates all four.

Assumption 1: the filter is sized for the number of keys it actually holds. Every figure above is a function of m/n, bits per key. Size the filter for 10 bits at a million keys and then put five million keys in it, and m/n falls to 2, at which point 0.6185 ^ 2 = 0.38 — a 38% false-positive rate, and the filter has stopped filtering. This is a real failure mode because SSTable sizes are not known before the file is written; implementations either estimate the key count or size the filter after the fact.

Assumption 2: the hash positions are independent and uniform. The whole derivation treats each of the kn bit-settings as an independent uniform draw. A weak or correlated hash clusters the bits, so fewer distinct positions are set and the measured false-positive rate exceeds the formula. The _positions method below uses the Kirsch-Mitzenmacher construction — two independent hashes combined as h1 + i*h2 — precisely because it produces k well-spread positions from one hash computation with no asymptotic loss.

Assumption 3: the filter is resident in RAM. The entire payoff is trading a disk read for a memory reference, so it holds only while all 2.34 GB per node stays in memory. If the filters are paged out under memory pressure, every lookup costs a disk read for the filter on top of the SSTable read it was supposed to avoid, and the mechanism inverts from a 122x saving into a penalty. This is why the growth row in Bottlenecks and scaling is about filter RAM rather than about disk.

Assumption 4: the page cache is cold. The page cache is the operating system’s in-memory copy of recently read disk pages, and in production it absorbs most reads of hot keys, so the 6.1-devices figure overstates steady-state need. That is an honest caveat rather than a retraction: 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. A filter also cannot forget — deleting a key cannot unset its bits without a counting variant — so a filter’s false-positive rate only ever drifts upward as an SSTable ages.

The listing below is a working filter plus the two formulas above. _positions is where the k probe positions come from; add and __contains__ are the two-line mechanism described earlier, with p >> 3 picking the byte and p & 7 picking the bit inside it.

The assertions are the part to read. The first two reproduce the table’s 0.82% and 0.046% rows. Then 200,000 keys go in, and two facts are checked against a real bit array: every key that was added is found (no false negatives), and the measured false-positive rate over 200,000 keys that were never added lands within 0.002 of the formula. The last two assertions reproduce the 0.0655 and 1.0574 figures from the block above.

import hashlib
import math


class BloomFilter:
    """m bits, k probe positions. False positives, never false negatives."""

    def __init__(self, n_keys, bits_per_key=10):
        self.m = n_keys * bits_per_key
        self.k = max(1, round(bits_per_key * math.log(2)))
        self.bits = bytearray((self.m + 7) // 8)

    def _positions(self, key):
        # Kirsch-Mitzenmacher: two independent hashes generate all k indices
        # with no asymptotic loss, so k is a loop bound, not k hash calls.
        d = hashlib.blake2b(key, digest_size=16).digest()
        h1 = int.from_bytes(d[:8], "big")
        h2 = int.from_bytes(d[8:], "big") | 1
        for i in range(self.k):
            yield (h1 + i * h2) % self.m

    def add(self, key):
        for p in self._positions(key):
            self.bits[p >> 3] |= 1 << (p & 7)

    def __contains__(self, key):
        return all(self.bits[p >> 3] >> (p & 7) & 1 for p in self._positions(key))


def fp_rate(bits_per_key, k=None):
    """(1 - e^(-k n / m))^k at the optimal integer k when none is given."""
    k = k or max(1, round(bits_per_key * math.log(2)))
    return (1 - math.exp(-k / bits_per_key)) ** k


assert abs(fp_rate(10) - 0.008194) < 1e-5      # 0.82% at 10 bits/key
assert abs(fp_rate(16) - 0.000459) < 1e-5      # 0.046% at 16 bits/key

bf = BloomFilter(200_000, bits_per_key=10)
for i in range(200_000):
    bf.add(f"key:{i}".encode())

assert all(f"key:{i}".encode() in bf for i in range(200_000))     # no false negatives
observed = sum(f"absent:{i}".encode() in bf for i in range(200_000)) / 200_000
assert abs(observed - fp_rate(10)) < 0.002, (observed, fp_rate(10))


def sstable_reads(n_sstables, holders, p):
    """Expected disk reads per lookup: real hits plus false positives."""
    return holders + (n_sstables - holders) * p


assert abs(sstable_reads(8, 0, fp_rate(10)) - 0.0655) < 1e-3     # absent key
assert abs(sstable_reads(8, 1, fp_rate(10)) - 1.0574) < 1e-3     # present in one

Digest reads

One more read-path saving is worth naming, and it is on the network rather than the disk. In a digest read the coordinator asks one replica for the full value and the rest for only a 32-byte hash of it, then requests full values only when the digests disagree.

At R = 3 that is 1,088 + 2 x 32 = 1,152 bytes on the wire instead of 3 x 1,088 = 3,264, so 1 - 1,152 / 3,264 = 0.647 — 65% off the read path’s network cost, for one extra round trip in the rare disagreeing case.

5. Hinted handoff and sloppy quorum

When a replica is unreachable, one mechanism keeps writes flowing — at the price of a guarantee, and at two very different rates that differ by a factor of a thousand and are routinely quoted for each other.

Strict versus sloppy

A strict quorum requires the W acknowledgements to come from the top N machines of the preference list. At N = 3, W = 2, one node being down is fine; two being down means the write fails.

A sloppy quorum drops that requirement. If s1 is unreachable, the coordinator walks past the top three to the next healthy machine, s4. That machine stores a hint — the value plus a note saying “this belongs to s1” — and replays it once s1 comes back. The write is acknowledged, and availability is preserved.

The whole mechanism, storing the write on a stand-in and handing it over later, is called hinted handoff.

The moment it fires, W + R > N stops being true

The diagram below is a four-step timeline that produces a stale read with no rule broken. Read it left to right and track which machines are in each set — that, not the arithmetic, is where it goes wrong.

At t0, s1 and s2 are both unreachable, so the write acked by s3 and s4 is all the write there is, and s4 holds a hint for s1. At t1, s1 and s2 return, and s3 is now slow. A read arrives and is answered by s1 and s2, which satisfies R = 2. The result is stale, because neither replica that answered the read was in the set that took the write.

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"]

    style T2 fill:#bc6c25,color:#fff
    style T5 fill:#9d0208,color:#fff

The writer set {s3, s4} and the reader set {s1, s2} share no members at all. The arithmetic W + R = 4 > 3 still held, and it bought nothing, because the pigeonhole argument in The quorum what w and r actually buy assumed both sets were drawn from the same N nodes — assumption 3 in that section, violated here by construction. Sloppy quorum trades away the exact guarantee that quorum existed to provide.

Two rates, a thousand apart

So how often does that happen? There are two different questions here, their answers differ by a factor of a thousand, and attaching the wrong number to the wrong question is the easiest mistake in this section to make.

The trigger for a sloppy write is one preference node unreachable: s1 is down, the coordinator walks to s4, and the write is now outside the preference list.

The trigger for an actual stale read is two unreachable. With W = R = 2, a writer set that still contains two preference nodes cannot possibly be disjoint from a two-node reader set — two 2-element subsets of a 3-element set always overlap. It takes a second node going down to shrink the writer set inside the preference list to one.

Assume 99.9% availability per node, so each node is unreachable with probability q = 0.001:

P(>= 1 of 3 unreachable): 1 - 0.999 x 0.999 x 0.999           =  0.002997
writes routed outside the preference list, /s
                          100,000 x 0.002997                  =  299.7        writes/s
per day:                  299.7 x 86,400                      =  25,894,089   writes/day

P(>= 2 of 3 unreachable): 3 x 0.001 x 0.001 x 0.999 + 1e-9    =  0.000002998
writes whose writer set can be disjoint from a later reader set, /s
                          100,000 x 0.000002998               =  0.2998       writes/s
per day:                  0.2998 x 86,400                     =  25,903       writes/day
ratio:                    0.002997 / 0.000002998              =  1,000

The two probability lines are worth reading slowly. The first is one minus “all three up.” The second counts “exactly two down” as 3 x 0.001 x 0.001 x 0.999 — the 3 is which node stayed up — plus 0.001 ^ 3 = 1e-9 for all three down.

About 25.9 million writes a day land outside their preference list, and about 26,000 of those are in a state where a later quorum read can miss them.

The 26,000 is the correctness number, and the one Failure modes alerts on. The 25.9 million is the number that sizes hint volume, and it is a thousand times larger.

Both are numbers, which is the entire point. You can decide whether 26,000 exposed writes a day is acceptable for your product. You can decide nothing whatsoever about the phrase “eventually consistent” — which, as Framing what decision and what breaks defined it, promises convergence only once updates stop and puts no bound on anything before that.

The block below is that arithmetic as code, plus a proof of the “one node down is not enough” claim. The two for down in combinations(...) loops at the bottom are the interesting part: they enumerate every possible down-set and check by brute force that one node down always leaves the writer and reader sets overlapping, while two nodes down does not.

from math import comb


def p_at_least(k, n, q):
    """P(at least k of n nodes are unreachable), each unreachable w.p. q."""
    return sum(comb(n, i) * q ** i * (1 - q) ** (n - i) for i in range(k, n + 1))


WRITES_PER_S, SECONDS_PER_DAY, Q = 100_000, 86_400, 0.001

# The trigger this section defines for a sloppy write: ONE node down.
sloppy = p_at_least(1, 3, Q)
assert abs(sloppy - 0.002997) < 1e-9
assert round(WRITES_PER_S * SECONDS_PER_DAY * sloppy) == 25_894_089

# The trigger for a writer set that can be disjoint from a reader set at
# W = R = 2: TWO down. This is the 2.998e-6 rate, not the 2.997e-3 one.
disjoint = p_at_least(2, 3, Q)
assert abs(disjoint - 0.000002998) < 1e-9
assert round(WRITES_PER_S * SECONDS_PER_DAY * disjoint) == 25_903

# The two are a factor of 1,000 apart, so the labels are not interchangeable.
assert round(sloppy / disjoint) == 1_000

# Why two, not one: enumerate the sets. With one preference node down the
# write still lands on two of them, and any two-node read set overlaps.
from itertools import combinations

pref = ("s1", "s2", "s3")
for down in combinations(pref, 1):
    writers = [p for p in pref if p not in down]          # plus a hint on s4
    assert all(set(writers) & set(readers)
               for readers in combinations(pref, 2))
for down in combinations(pref, 2):
    writers = [p for p in pref if p not in down]
    assert any(not (set(writers) & set(readers))
               for readers in combinations(pref, 2))

Hint volume, and the storm it causes

Hint volume is the second cost, and it is the one that takes nodes down. For an outage of one node, at the 18,750 writes/s destined for it:

10-min outage:        18,750 x 600                        =      11,250,000  hints
hint bytes:           11,250,000 x 1,088                  =  12,240,000,000  B  = 12.24 GB
replay at 125 MB/s:   12,240,000,000 / 125,000,000        =            97.9  seconds
3-hour outage:        18,750 x 10,800                     =     202,500,000  hints
hint bytes:           202,500,000 x 1,088                 = 220,320,000,000  B  = 220 GB
replay:               220,320,000,000 / 125,000,000       =           1,763  seconds = 29 min

A ten-minute outage accumulates 12.24 GB of hints and 98 seconds of replay — and that replay arrives on top of the 20.4 MB/s of live traffic the recovering node is already accepting.

That is the hint storm: the node returns, every peer floods it with buffered writes at once, it falls over again, and yet more hints accumulate behind it. The mitigations are to rate-limit the replay, and to rejoin the ring for writes before rejoining it for reads.

At Cassandra’s default three-hour hint window the same arithmetic gives 220 GB and 29 minutes of replay at full line rate. No node buffers that on behalf of a peer, which is exactly why the window exists.

Past that window hints are dropped, and the only remaining path back to agreement is the Merkle comparison in Merkle trees one bad key in a million. Hinted handoff and anti-entropy repair are not competing alternatives. They are the fast and the slow path of the same job, and the hint window is the boundary between them.

Read repair, the third mechanism

Read repair is the cheapest of the three. When a quorum read finds the replicas disagreeing, the coordinator writes the merged value back to the stale ones. That is the repair list returned by quorum_read in Vector clocks and the sibling nobody wants, which covers every replica that answered, not only the R the read waited for.

It costs no extra requests, because the responses were already paid for, and it fixes exactly the keys people are actually reading.

That last point cuts both ways. Real access patterns follow a Zipf distribution, meaning a small number of keys absorb most of the traffic. Read repair fixes that hot head and does nothing at all for the cold tail — which is what scheduled anti-entropy is for.

And read repair is blind to one case by construction. Two values carrying the same clock look identical to descends, so repair sees nothing to fix, and both values must simply be kept as siblings (Vector clocks and the sibling nobody wants).

What hinted handoff assumes, and what fails when it does not hold

It assumes outages are short relative to the hint window. Past three hours the hints are discarded outright, and a cluster that is not running scheduled repair will then never converge. The divergence is permanent and completely silent.

It assumes the stand-in node has spare disk and spare bandwidth, which during a correlated incident it does not, because every peer is buffering for the same dead node at once.

It assumes the recovering node can absorb the replay, which the hint-storm arithmetic above shows it frequently cannot without rate limiting.

And it assumes the failure detector is right. A node wrongly marked down still receives no writes and accumulates hints it never needed, so a flapping node generates hint volume proportional to how often it flaps. That is what makes Membership gossip and phi accrual part of this mechanism rather than a separate topic.

6. Membership: gossip and phi-accrual

All of the machinery above depends on knowing who is alive, with no central registry to ask — and the decision “is this node down?” should produce a number rather than a yes or no.

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 the way a rumour does, so the whole fleet converges on a consistent view in a number of rounds proportional to the logarithm of the fleet size, written O(log S) — about four rounds at S = 16, and only about ten at a thousand nodes.

That part is routine. The interesting part is the failure detector, the component that decides a node is dead, because What v costs memory lookup gossip and durability showed that a flapping node — one that alternates between healthy and unreachable — forces the cluster to move data in and back out on every transition.

Why a fixed timeout cannot win

Make the timeout short, and a garbage-collection pause evicts a perfectly healthy node. Make it long, and a genuinely dead node keeps being sent writes. There is no setting that is right in both cases, because 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, a small “I am alive” message, on a regular interval. The detector records the gaps between arrivals, fits a probability distribution to them, and reports:

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

Read that as: the negative base-10 logarithm of the probability that a healthy node would have stayed silent this long. A phi of 8 means “if this node were healthy, a silence this long would be a one-in-a-hundred-million event,” because 10 ^ -8 is one in a hundred million.

Work the threshold out for the simplest case. For exponentially distributed gaps with a 1-second mean, P(> t) = e^(-t), so:

phi(t)  =  -log10(e^(-t))  =  t / ln(10)  =  t / 2.3026
phi = 8 fires at:     8 x 2.3026                          =  18.4  seconds of silence

So the detector declares a node down after 18.4 seconds of silence with a 1-second heartbeat — and, the property a fixed timeout cannot have, that threshold stretches automatically when the network slows, because the fitted distribution widens along with the observed gaps.

Two follow-ups are worth volunteering.

First, the output is a suspicion level, not a verdict, so different subsystems can pick their own thresholds. Stop routing reads to a node at phi = 4, while ring membership waits for phi = 8 plus a long timeout.

Second, “down” must never automatically mean “removed from the ring.” Removing a node redistributes 1/S of the entire corpus — 2.04 TB of movement here (ch 05). Decommissioning a node, meaning permanently taking it out of the ring and rehoming its data, stays a human decision.

7. CAP, stated for this store

What does all of the above add up to during a network partition? Start with what the theorem actually says, because it is usually misquoted.

CAP stands for Consistency, Availability, and Partition tolerance. The theorem is a statement about what a system does while a network partition is in progress — a partition being a failure in which two groups of machines are both running but cannot reach each other.

It is not a menu you pick two items from. Partitions are imposed on you by the physical world, so P is not something you choose. What you choose is how the reachable side behaves while one is happening.

Two corollaries follow immediately. 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 at all.

PACELC is the extension that covers that second case: during a Partition you trade Availability against Consistency; Else — when the network is fine — you trade Latency against Consistency. The full statement, the four things people routinely get wrong, and PACELC in detail are in Cap and the framing that is actually useful. Cite them rather than re-deriving them.

Applying it to this store

What is specific to this store is that W is the knob, and it is set per request rather than per cluster.

Suppose a partition splits the preference list into {s1, s2} on one side and {s3} on the other. The table below runs three values of W through that split. In the classification column, PC means the system chose consistency over availability during the partition, and PA means it chose availability over consistency.

Pricing the W = 1 choice

Take a 60-second partition with the traffic split evenly between the two sides, and count how many keys end up written on both sides and therefore become siblings:

writes in 60 s:       100,000 x 60                            =  6,000,000  writes
each side:            6,000,000 / 2                           =  3,000,000  writes per side
keys written on both: 3,000,000 x 3,000,000 / 10,000,000,000  =        900  keys
sibling rate:         900 / 6,000,000                         =    0.00015  = 0.015%

The third line is the only one that needs explaining. Pick one of the 3,000,000 keys written on the left. The chance a given right-side write hits that same key is 1 / 10,000,000,000, since there are 10 billion keys. With 3,000,000 right-side writes, the expected number of collisions is 3,000,000 x 3,000,000 / 10,000,000,000 = 900.

That is 900 conflicting keys out of 6 million writes, or 0.015% — which lines up with the figure Amazon reported for Dynamo in production: 99.94% of requests saw exactly one version.

And that is the honest defence of W = 1. Conflicts are rare because 10 billion keys is an enormous space, and two writers landing on the same key in the same minute is unlikely.

The defence assumes writes are spread uniformly across the key space, and it dies under a Zipf distribution. With a small hot set absorbing most of the traffic, collisions concentrate on those keys and the count rises by orders of magnitude. That is the same skew What consistent hashing does not fix declines to solve with hashing, and it is the assumption to state out loud whenever you quote the 0.015%.

3. Bottlenecks and scaling

What actually runs out first depends on the operating regime, and each regime has a different answer.

Two abbreviations appear in the table below: QPS is queries per second, how many requests the system handles each second, and gc_grace is Cassandra’s setting for how long a tombstone is retained before compaction is allowed to discard it.

Read the middle column first — it is the resource that binds in each regime, and every one of those numbers was derived above. Notice that four of the seven rows bind on bandwidth, not CPU or disk space.

Running across multiple regions is where the quorum choice stops being subtle.

With the three copies spread one per region, a W = 2 write must wait for a second region to answer, so it cannot finish faster than one cross-region round trip. That is about 70 ms, against 0.5 ms within a region:

cross-region penalty:  70 / 0.5                           =  140x the local cost

Every write pays 140 times the local cost, and those extra 70 ms sit on the user’s critical path rather than in a background job.

The standard answer is to keep three copies per region and use LOCAL_QUORUM, a Cassandra consistency level meaning “count acknowledgements only from replicas in the caller’s own region.” You then accept two things: cross-region convergence is asynchronous, and a regional partition produces siblings.

This is the “Else” branch of PACELC — the latency-versus-consistency trade you pay on every request when the network is healthy — and you spend 99.99% of your life in it.

4. Failure modes

Each row below is one way the design breaks in production: a concrete trace of how the failure unfolds, the signal that would let you notice it, and the guard that prevents it.

The theme running through the table is that the dangerous failures are the silent ones. A stale read, a truncated clock, and a resurrected tombstone all produce wrong answers with no error anywhere — which is why the Detection column matters more than the Guard column.

5. Alternatives rejected

A reasonable person would propose several other designs instead. Each is genuinely good at something, and each is ruled out here by a specific piece of arithmetic or a specific requirement.

One term in the table needs a definition first. A CRDT is a conflict-free replicated data type: a data structure whose merge operation is defined so that 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 middle column is the one to read in an interview — naming what an alternative is genuinely good at is what makes the rejection credible.

The one to revisit, with a trigger: move to per-key consensus the first time an application team asks you “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 quietly lose data forever.

6. Interviewer pushback

Nine questions this design attracts, each with the answer written out the way you would say it aloud, and a note on what the question is really testing.

Everything in them is derived above. What is new is the phrasing and the order the numbers come out in.

“Why W + R > N? Prove it.” Testing: whether you can say the pigeonhole argument. The write landed on some W-subset of the N replicas and the read consults some R-subset. Two subsets of an N-set of sizes W and R must share at least W + R - N members, so W + R > N forces the intersection to be non-empty and the read set is guaranteed to contain at least one replica holding the last completed write. At N = 3, W = R = 2 the guaranteed overlap is exactly 2 + 2 - 3 = 1. And I would immediately add the caveat: this is a statement about the last completed write. It does not order concurrent writes, it does not roll back a partial write, and it is void under sloppy quorum.

“So which (W, R) do you pick, and what does it cost?” Testing: whether latency and availability are numbers to you. (2, 2) by default. If one replica exceeds 10 ms with probability 1%, then waiting for the second of three exceeds it with probability 3(0.01)^2(0.99) + (0.01)^3 = 0.0298% — 34 times better than a single replica, because I discard the slowest of three. Waiting for all three gives 1 - 0.99^3 = 2.97%, 3x worse than one replica. Availability is even more lopsided: at 99.9% per node, W = 3 blocks writes 0.2997% of the time and W = 2 blocks 0.0003%, a factor of 1,000. W = 3, R = 1 is the read-optimized configuration — reads are as fast as one node because every replica has everything — and W = 1, R = 3 is its mirror. You are choosing which path eats the third replica’s tail, not choosing between fast and slow.

“What does W = 1, R = 1 actually give up?” Testing: whether “eventual consistency” is a phrase or a quantity. The overlap guarantee, and here is the number: the write acks after one replica has it, so a read issued inside the replication window returns the new value only if it happens to land on that replica — at most one chance in three under a random-replica model, and better than that in practice because R = 1 takes the fastest answer and the replica that acked first tends to answer first, so up to two-thirds of reads in that window are stale. Normally the window is sub-millisecond and nobody notices. It is unbounded during a GC pause or a partition, and nothing in the protocol reports it. (1,1) is correct except when something is wrong, which is the profile of a bug you find during an incident and not before.

“Walk me through two clients writing the same cart at the same time.” Testing: whether you can trace a vector clock rather than name it. Client A writes {milk} through coordinator Sx, producing clock [(Sx,1)]. Client B writes {eggs} through Sy, producing [(Sy,1)]. Neither descends from the other — A’s clock has Sx = 1 > 0 and B’s has Sy = 1 > 0 — so they are concurrent and both are kept as siblings. The next read returns both values plus the pointwise-max context [(Sx,1),(Sy,1)]. The application knows a cart is a set, unions them, and writes back with that context, producing [(Sx,2),(Sy,1)], which dominates both siblings, so they are discarded. The store never resolved anything — it refused to guess and made the conflict visible.

“Vector clocks grow. What do you do about it?” Testing: whether you know the failure mode of the standard fix. They gain an entry per distinct coordinator that has ever written the key, so with coordinator failover the ceiling is the fleet size. At 24 bytes an entry, ten entries is 240 B on a 1,088 B record — plus 19.9%, and 23% of the 1 KB payload it describes. Dynamo caps at 10 and evicts the oldest by timestamp, and that truncation is lossy in a way that is not always safe: if clock {S1:1, S2:1} is truncated to {S1:1}, then a genuinely concurrent {S1:1, S3:1} now dominates it, and the first write is discarded with no sibling and no error. The real fix is to keep the clock server-side and per-replica so it is bounded by N = 3 rather than by the fleet, and to use dotted version vectors so a client writing blind in a loop does not manufacture siblings.

“Two replicas hold a million keys each. Find the one that differs.” Testing: whether O(log n) is a shape or an arithmetic. Merkle trees. Hash each key into a leaf, each internal node into the hash of its children. Exchange roots first: 32 bytes, one round trip, and if they match the whole million keys are proven identical — that is the common case and it is why this is cheap. If they differ, descend: at each level compare the two children of the node you know is bad, so 1 + 2 x 20 = 41 hashes and 1,312 bytes localises the key, against 1,048,576 comparisons or 1.14 GB of streaming. Round trips are the separate question: one message per level is 21 round trips, about 10.5 ms at a 0.5 ms intra-datacenter RTT, and shipping the whole tree is 67 MB. So ship the top 11 levels in one message and the surviving subtree in a second — 2 round trips, 131 KB, 8,708x less than streaming. The honest cost is that building the tree is a full sequential read: 2.04 TB at 1 GB/s is 34 minutes per node, 9.1 hours for a 16-node fleet, which is why repair is a weekly scheduled job.

“Why does an LSM point read need a bloom filter?” Testing: the arithmetic, not the concept. Because with size-tiered compaction a key can be in any of ~8 SSTables and is in at most one, so 7 of every 8 lookups are guaranteed misses. At m/n bits per key with k probes the false-positive rate is (1 - e^(-kn/m))^k, minimized at k = (m/n) ln 2. At 10 bits per key that is k = 6.93, so 7, and the rate is (1 - e^(-0.7))^7 = 0.82%. Expected disk reads for an absent key 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 devices: a 100 MB/s random device does 100e6 / 4,096 = 24,414 reads/s, and this node needs 18,750 read requests/s, so without filters that is 150,000 IOPS, which is 6.1 devices, and with them 19,826, or 0.81 of one. The closed form is 0.6185^(m/n), so every 4.8 extra bits per key divides the FP rate by ten, and the RAM cost is 2.34 GB per node at 10 bits over 1.875 billion keys.

“A node is down and you keep accepting writes. What did that cost?” Testing: whether “highly available” has a price tag attached. Sloppy quorum sends the write past the preference list to the next healthy node, which stores a hint. The cost is exactly the guarantee quorum was for: the writer set and the later reader set can be disjoint, so W + R > N holds arithmetically and buys nothing. Two rates, and they are a thousand apart, so I would be careful which one I quoted. Landing outside the preference list needs one node unreachable, which at 99.9% per node is 1 - 0.999^3 = 3.0e-3 and about 25.9 million writes a day. Being exposed to a stale read needs two unreachable, because with W = R = 2 a writer set holding two preference nodes still overlaps every reader set — that is 3(0.001)^2(0.999) = 3.0e-6, about 26,000 writes a day. The 26,000 is the correctness number; the 25.9 million is what sizes the hints. Then there is volume: a 10-minute outage accumulates 18,750 x 600 = 11.25 M hints, 12.24 GB, 98 seconds of a saturated NIC to replay on top of live traffic — the hint storm that knocks the recovering node over a second time. Past the 3-hour hint window it would be 220 GB, so hints are dropped and Merkle repair becomes the only path back to convergence.

“Is this AP or CP?” Testing: whether you say “pick two.” Neither, as a property of the system — it is a per-request setting, and CAP only describes behaviour during a partition. If a partition splits the preference list into {s1,s2} and {s3}, then W = 2 makes the minority side refuse writes, which is CP; W = 1 lets both sides accept, which is AP and produces siblings. And the partition branch is the boring one: partitions are rare and the latency-versus-consistency trade is paid on every single request, which is the E in PACELC. Concretely, a 60-second partition at 100,000 writes/s with traffic split evenly produces about 3e6 x 3e6 / 1e10 = 900 keys written on both sides — 0.015% of the writes — which is the honest defence of W = 1 and also its limit, because a Zipf distribution concentrates the writes and blows that estimate up. One thing I would not say: a single-node store is not “CA.” It is not a distributed system, so the theorem does not apply to it.

Cheat sheet

Every result in the chapter, compressed to one line each, in the order they were derived.

Next: 07 — Design A Unique ID Generator — where the key this store is partitioning by has to come from somewhere, and 64 bits is a budget you have to spend deliberately.