Design a key-value store

A key-value store does one thing: given a key, it stores or returns a blob of bytes. Behind that tiny API sit some of the most reused ideas in distributed systems. In this problem you design a Dynamo-style store — leaderless, spread over 120 machines, every value kept three times — for data that must always be writable: shopping carts during a Big Billion Days–style sale, login sessions, user profiles, the device-binding record a UPI app checks before every payment. The API is the easy part. The interesting parts are how you spread 100 TB so that adding a machine moves almost nothing, how three copies can disagree and still converge, what “consistent” even means when any replica can take a write, and how one machine absorbs 7,500 replica writes a second without its disk falling behind.

Understand the problem

Other teams will build on this store, so pin down what “never lose” means and what happens when the network breaks. This is the system of record: a cache may forget a value and refetch it, this store may not. For the volatile version, see design a distributed cache.

Functional requirements

1. Clients should be able to put, get and delete a value by key. Keys are up to 256 bytes; values up to 1 MB, most a few KB.
2. Clients should be able to choose consistency per request — ONE, QUORUM or ALL — trading latency for freshness on each call.
3. Clients should be able to set a time-to-live on a key, so sessions and short-lived records expire on their own.
4. The system should never silently drop a concurrent update. When two writes race, a read returns both versions (or one merged value) along with a context the client sends back on its next write.

Non-functional requirements

• Always writable. Writes are accepted through a node failure, a zone outage or a network partition. Availability target for the request path: 99.99% a month — about 4.3 minutes of errors (43,200 minutes × 0.0001).
• Low, predictable latency. Within one region, p50 under 2 ms and p99 under 10 ms for gets and puts of values up to 10 KB.
• High throughput. 1M gets/s and 300k puts/s at peak (deletes count as puts); a normal day runs at about a third of that.
• Scale. 50 billion keys, average item 2 KB, so 100 TB of logical data, growing by about a third a year.
• Durability. An acknowledged write survives the loss of any one machine or any one zone.
• Elastic. Nodes can be added, removed or replaced with no downtime and without breaking the latency targets.

Out of scope

• Secondary indexes and queries by value. You can only look things up by key.
• Multi-key transactions and SQL. Every operation touches one key; atomic updates across keys belong to a transactional database (see ACID transactions).
• Range scans. Hashing keys destroys their order; an ordered layout is discussed as a follow-up.
• Large objects. Anything over 1 MB goes to blob storage, with only a pointer stored here.
• Multi-region replication is a follow-up, not part of the core design.

The set-up

Plan the approach

Start with one machine, then partition, then replicate, one functional requirement at a time. Say the shape early: “every node is identical and can coordinate any request — no leader. The only special component is a small metadata store holding the partition map.” Then spend most of the time in the deep dives.

Core entities

• Key. A keyspace (a namespace such as carts) plus the key bytes, hashed to a 64-bit token.
• Versioned value. The bytes plus a version vector, a write timestamp, an optional expiry time and a tombstone flag for deletes.
• Partition. One of 4,096 equal slices of the token space. It is the unit of placement, replication, repair and movement.
• Replica set (preference list). The three nodes, one per availability zone, that hold a partition, plus fallback nodes for writes when one of them is down.
• Node. A host ID, its zone, the partitions it holds and its liveness as seen through gossip.
• Hint. A write held on behalf of a replica that was down: the target node, the mutation and its time.

Each keyspace carries its own settings, because carts, sessions and device bindings want different trade-offs:

keyspace: carts
replication: { factor: 3, placement: one-per-zone }
consistency: { read: QUORUM, write: QUORUM }   # callers may override per request
conflicts: siblings          # lww | siblings | crdt:or-set | cas
default_ttl: 30d
compaction: leveled          # leveled | tiered
gc_grace: 10d                # how long delete markers are kept

API

Production clients use a binary protocol over long-lived connections; HTTP is shown for readability. The context is an opaque token carrying the version vector of what the client last read. Sending it back on a write says “this replaces what I saw”.

GET /v1/kv/carts/user:98231?consistency=QUORUM

{
  "key": "user:98231",
  "versions": [
    { "value_b64": "eyJpdGVtcyI6WyJkYWwiXX0=", "expires_at": "2026-11-26T10:00:00Z" },
    { "value_b64": "eyJpdGVtcyI6WyJhdHRhIiwicmljZSJdfQ==", "expires_at": "2026-11-26T10:00:04Z" }
  ],
  "context": "eyJBIjoyLCJCIjoxfQ==",
  "replicas_answered": 2
}

Two versions came back: the cart had two concurrent edits. The client merges them and writes the result with the context it received:

PUT /v1/kv/carts/user:98231?consistency=QUORUM&ttl=2592000
Content-Type: application/octet-stream
X-KV-Context: eyJBIjoyLCJCIjoxfQ==

{"items":["atta","dal","rice"]}

{ "ok": true, "context": "eyJBIjozLCJCIjoxfQ==", "replicas_acked": 2 }

A delete is a write too: DELETE /v1/kv/carts/user:98231 with the context writes a tombstone. Errors are honest about uncertainty. If fewer than W replicas acknowledge before the timeout, the response is 503 with replicas_acked: 1 — and the write may still exist on that one replica and spread later, so clients must treat a failed put as “unknown”, not “didn’t happen”. Values over 1 MB get 413.

Back-of-envelope estimates

Show every step; these numbers set the node count, the memory budget and the compaction strategy. (For the method, see back-of-envelope estimation.)

Storage. 50 billion keys × 2 KB = 100 TB logical. Three replicas: 100 × 3 = 300 TB. An LSM tree with leveled compaction needs about 1.15× that for not-yet-compacted versions (RocksDB’s worked example comes to 1.14), so 300 × 1.15 = 345 TB on disk.

Replica operations. A quorum read asks two replicas: 1M × 2 = 2M replica reads/s. A write goes to all three: 300k × 3 = 900k replica writes/s. Total 2.9M replica operations/s.

Nodes. Budget 40k replica operations/s per node, keeping CPU near 50% so p99 stays flat: 2.9M ÷ 40k = 72.5, so 73 nodes. For storage, give each node an 8 TB NVMe drive kept at most half full (compaction and streaming need room): 345 ÷ 4 = 86.25, so 87 nodes. Now survive a zone outage: with three zones, losing one leaves two-thirds of the fleet, so you need 73 × 3 ÷ 2 = 109.5 nodes. Round to 120 nodes, 40 per zone. Each holds 345 ÷ 120 ≈ 2.9 TB, about 36% of its disk.

Write bandwidth per node. 300k puts/s × 2 KB × 3 replicas ÷ 120 nodes = 15 MB/s of new data per node. If compaction rewrites each byte about 33 times (see the storage-engine deep dive), that becomes 15 × 33 ≈ 500 MB/s of background disk writes at peak — the number that decides the compaction strategy.

Memory. Keys per node: 50 billion × 3 ÷ 120 = 1.25 billion. Bloom filters at 10 bits per key: 1.25 billion × 10 bits = 12.5 Gbit ≈ 1.6 GB per node. Add a few GB of memtables; the rest of a 128 GB machine is block cache.

Network. Reads return 1M × 2 KB = 2 GB/s and writes ship 300k × 2 KB × 3 = 1.8 GB/s: about 3.8 GB/s ÷ 120 ≈ 32 MB/s per node, trivial for a 25 Gbit/s NIC. Repair and streaming are the network budget that matters.

Hints. If one node is down, the writes it misses are its share of replica writes: 900k ÷ 120 = 7,500/s × 2 KB = 15 MB/s. A 10-minute restart leaves 15 MB/s × 600 s = 9 GB of hints on other nodes.

High-level design

Build the store one requirement at a time. By the end of this section it works in one region at moderate load; the deep dives make it survive scale and failure.

Clients should be able to put, get and delete by key

On one machine this is a hash map backed by a log. At 100 TB it must be partitioned. Hash each key to a 64-bit token (Dynamo used MD5, Cassandra uses Murmur3) on a ring that wraps around; each node owns several positions on it. To find a key’s replicas, walk clockwise from its token and collect the first three distinct machines, in three distinct zones. That ordered list is the key’s preference list.

Any node can coordinate a request, and every node knows the whole map, so a request reaches the right replica in one hop — Dynamo called this a zero-hop distributed hash table. Better, ship a partition-aware client library that caches the map and sends each request straight to a replica. Dynamo measured that switch: 99.9th-percentile read latency fell from 68.9 ms to 30.4 ms and the average from 3.9 ms to 1.55 ms, because the load balancer and the extra hop disappeared.

The flow is short:

1. The client hashes carts/user:98231 to a token and looks up its preference list in the cached map: B, D, C.
2. It sends the request to B, the first healthy replica, which becomes the coordinator.
3. B writes or reads its local copy and talks to D and C, as the next requirement describes.
4. B answers the client. A delete writes a tombstone rather than erasing anything; the storage-engine deep dive explains why.

Clients should be able to choose consistency per request

Three numbers describe every request. N is the number of replicas (3). W is how many must acknowledge a write before the coordinator reports success. R is how many must answer a read. The coordinator always sends a write to all N replicas; W only decides how long it waits. If R + W > N, every read set shares at least one replica with every write set, so a read sees the latest acknowledged write — as long as nothing has failed.

Callers choose per request from a small menu, the same one Cassandra exposes so callers never need to know N:

To save bandwidth, a quorum read asks one replica for the full value and the others only for a hash of it; if the hashes disagree, the coordinator fetches the full values and repairs the stale replica.

Clients should be able to set a time-to-live

Store an absolute expiry time with each version, computed once by the coordinator, so every replica agrees when the value dies even if it receives the write an hour late through a hint. Reads treat expired versions as missing, and compaction removes them later. Milliseconds of clock skew don’t matter for TTLs of minutes or days. Mostly-TTL data, such as 30-minute sessions, also shapes compaction: files that expire together can be dropped without being rewritten.

The system should surface concurrent updates

Any replica can accept a write, so two clients can update one cart at once through different coordinators, or on opposite sides of a partition. The store must notice rather than keep one version at random. Dynamo’s answer is a version vector (it says vector clock): a small map from coordinator node to counter, stored with every version. If every counter in version X is less than or equal to the matching counter in Y, then Y descends from X and X can go. If neither covers the other, the writes were concurrent, and the store keeps both as siblings.

So the API returns a list of versions and a context. The client merges the siblings and writes with the merged context, [A:2, B:1]; the coordinator bumps its own counter to [A:3, B:1], which covers both branches, so the siblings collapse into one. Conflicts are rare: over 24 hours, 99.94% of requests to Dynamo’s cart service saw exactly one version, and most divergence came from busy automated clients writing concurrently, not from failures.

You now have a working store. The deep dives cover what breaks at 120 nodes and 300k writes a second.

Deep dives

How do we partition data across nodes?

Partitioning decides whether load is even and how much data moves when the cluster changes. Moving data is the expensive part: every byte is read from one disk, sent over the network and written to another while the cluster keeps serving a sale’s traffic.

Bad: hash(key) mod N. Simple and perfectly even, until N changes. Going from N to N + 1 nodes changes the owner of N/(N + 1) of all keys. At 120 → 121 nodes that is 99.2% — about 342 of the 345 TB on disk reshuffled to add one machine.

Good: consistent hashing, one token per node. Each node sits at one random point on the ring and owns the arc back to its predecessor. A newcomer takes part of one neighbour’s arc, so adding the 121st node moves about 1/121 of the data: 345 ÷ 121 ≈ 2.9 TB instead of 342 TB. But random points make uneven arcs, all the moved data comes from one neighbour, and a failed node dumps its whole load on its successor.

Better: virtual nodes. Give each machine many tokens. A newcomer takes small slices from every machine, a failed node’s load spreads across the cluster, and bigger machines hold more tokens. Cassandra 2.x defaulted to 256 random tokens per node, and its documentation lists the costs: each token adds up to 2 × (RF − 1) ring neighbours, so more pairs of failures make some range unavailable, and range-by-range repair slows down. Dynamo ran random tokens in production and found two more: a joining node’s data had to be found by scanning donors’ storage at low priority, which in the holiday season took almost a day, and every membership change reshaped ranges, forcing Merkle trees to be rebuilt.

Great: fixed partitions, placed independently. Dynamo’s final strategy splits the hash space into Q equal partitions that never change and assigns whole partitions to nodes, about Q ÷ S each for S nodes. Which keys belong together is now separate from which machines hold them. With Q = 4,096, each partition holds 100 TB ÷ 4,096 ≈ 24 GB of logical data (about 28 GB on disk), and each of 120 nodes holds 4,096 × 3 ÷ 120 ≈ 102 partition replicas.

A partition is its own set of files, so moving it is a file copy, not a scan, and its Merkle tree survives membership changes. Dynamo reported that this strategy balanced load best of the three it tried and cut membership metadata by three orders of magnitude. The price is coordination: two nodes must never both believe they own partition 17, so the map lives in a small, strongly consistent store such as etcd or ZooKeeper, off the data path and cached everywhere. And Q is fixed up front; if the cluster grows tenfold, split every partition in two (Q = 8,192) to keep the balance.

Hashing spreads keys, not requests to one key: a viral key still lands on three machines (a follow-up below). More in consistent hashing and database sharding.

How do we replicate and choose consistency?

N, R and W trade four things against each other: latency, availability, durability and freshness.

Latency is set by the slowest member of the quorum. Dynamo puts it plainly: a get or a put is only as fast as the slowest of the R or W replicas it waits for. Suppose three replicas answer in 1.2, 1.8 and 9.0 ms because the third is in a garbage-collection pause.

Two consequences follow. Ask one more replica than you need when the first ones are slow: Cassandra sends a speculative read to an extra replica after a short delay, and since writes already go to all three, W = 2 gets this for free. And prefer fast replicas: Dynamo makes whichever replica answered the preceding read fastest the coordinator of the write, which trims the tail and helps read-your-writes.

Dynamo’s users mostly ran (N, R, W) = (3, 2, 2); a few read-heavy services, such as a product catalogue, used R = 1 and W = N. With W = 1 a write is refused only if no node can take it, but it returns while one machine holds the value — a durability window to accept knowingly.

What “strong” means here — and what it does not. R + W > N buys overlap. It does not buy linearizability, for five reasons:

1. Sloppy quorums. During failures, writes land on stand-ins outside the preference list, which a normal read quorum may not include.
2. Failed writes are not rolled back. A put that reached one replica and then timed out returns an error, yet the value exists and may spread later.
3. Concurrent writes have no agreed order; last-write-wins orders them by timestamps that may disagree with real time.
4. Reads racing a write can see the new value and then the old one — unless the coordinator writes the newer value back to a quorum before replying. Do that; it costs a write only when replicas disagree.
5. Membership changes. While a partition streams to a new node, writes must go to old and new owners, or quorums stop overlapping.

For operations that need exactly one winner — “bind this UPI account to this phone only if no other phone is bound” — use compare-and-set backed by consensus on that partition. Cassandra offers it through lightweight transactions; DynamoDB runs every partition as a Multi-Paxos group whose elected leader serves all writes and strongly consistent reads. The price is extra round trips and a pause when a leader fails: a new DynamoDB leader waits a couple of seconds for the old lease to expire. Keep that path for keyspaces that need it (consensus and leader election).

How do we resolve conflicting writes?

A leaderless store will see conflicts: two coordinators accept writes to one key at once, or both sides of a partition stay writable. Who decides the outcome, and what gets lost?

Bad: last-write-wins on wall-clock time. Stamp each write with the coordinator’s or client’s clock; the larger timestamp wins everywhere. It needs no client code and always converges, which is why Cassandra uses it per column — with a warning in its documentation to keep clocks synchronised with NTP. It fails in two ways.

• Clock skew reorders writes. If node X’s clock runs 150 ms fast, its writes beat any write stamped up to 150 ms later. The later write vanishes.
• Concurrent writes lose data even with perfect clocks. The phone adds dal and the laptop adds rice 3 ms apart; one whole cart wins and the other add is gone.

Last-write-wins suits data where the newest whole value is the answer and losing one of two near-simultaneous writes is harmless; Dynamo’s session-state service used it. It is wrong for carts, counters and anything that accumulates. More in clocks, ordering and conflict resolution.

Good: version vectors with siblings, as in the high-level design. No acknowledged write is discarded silently, and causally later writes still replace earlier ones. You pay four ways:

• Every client, in every language, must merge siblings correctly.
• A naive union brings deleted items back, as the cart figure showed.
• Vectors grow when many nodes coordinate writes to one key. Dynamo keeps writes on the top N replicas and drops the oldest entry once a vector reaches about 10, accepting that a descendant may then look concurrent.
• Hot keys with many concurrent writers collect the most siblings.

Great: CRDTs — the merge rule lives in the data type. A conflict-free replicated data type has a merge that is commutative, associative and idempotent, so replicas can merge in any order, any number of times, and agree. The server merges; clients never see siblings. For a cart, use an observed-remove set: each add carries a unique tag, and a remove deletes only the tags it has seen. The phone’s remove deletes the original atta’s tag; the laptop never re-added atta, so it stays gone — but had the laptop added atta again, that new tag would survive, as the user meant. For counters, such as units reserved in a flash sale, keep one count per replica and sum them. Even last-write-wins is a CRDT — Cassandra calls its per-row rule an LWW-element-set — which is why it converges, just not always to what users expect.

The costs: metadata to garbage-collect (tags, per-replica counts, markers for removed tags), and only data with a sensible merge qualifies. A CRDT can never say “exactly one of you wins”.

When no merge is acceptable, refuse to diverge. Two phones racing to bind the same UPI account must not both succeed, and merging them means nothing. Use compare-and-set over per-partition consensus for that keyspace, and accept that it is unavailable on the minority side of a partition — the CP choice in CAP and PACELC.

How do we handle node failures?

At 120 nodes something is always broken — a disk, a network card, a kernel upgrade, a long GC pause; Dynamo’s authors called failure “the normal case”. Handle it in four layers: detect, write around, repair on read, repair in the background.

Detect: gossip plus an accrual failure detector. Every second, each node bumps its heartbeat counter and swaps its view of the cluster with one random peer — and sometimes a seed node, so the cluster never splits into islands — the rhythm Dynamo and Cassandra both use. News reaches all 120 nodes in roughly log₂ 120 ≈ 7 rounds (gossip protocol).

Deciding a node is down is the hard part. A fixed timeout is too slow on a calm network and too twitchy on a jittery one. Phi accrual detection (Hayashibara et al.) adapts: each node keeps a window of recent heartbeat gaps and reports a suspicion level φ = −log₁₀(1 − F(t)), where t is the time since the last heartbeat and F the distribution fitted to those gaps. φ = 8 means “if this node were alive, a silence this long would happen about once in 10⁸”. Convict above a threshold: 8 is a common default, 12 on noisy cloud networks.

Two rules keep detection safe. Each node decides for itself and acts locally, routing reads elsewhere and storing hints; Cassandra deliberately does not gossip those verdicts. And down is not gone: Dynamo and Cassandra remove a node from the ring only on an explicit command, so a reboot never triggers terabytes of rebalancing. A false alarm here costs a few hints, not a leader election. DynamoDB, which has leaders, found false alarms costly enough that a follower first asks its peers whether they can still reach the leader — a defence against gray failures. More in heartbeats.

Write around temporary failures: sloppy quorum and hinted handoff. When a replica is unreachable, the coordinator sends its copy to the next healthy node on the ring, tagged with a hint naming the intended owner. In Dynamo the stand-in keeps hinted copies in a separate local store and hands them over when the owner recovers; in Cassandra the coordinator keeps hints on disk for up to 3 hours of a replica’s downtime by default. Hints carry the original timestamp, so replay is idempotent and never overwrites a newer write.

A 10-minute restart leaves 9 GB of hints; a 3-hour outage leaves 15 MB/s × 10,800 s = 162 GB. Replay must be throttled or the returning node is flattened. Cassandra’s documentation works an example in which a 10-minute restart at 100 Mbit/s of ingest builds about 7 GiB of hints that take two hours to deliver at the default 1,024 KiB/s throttle.

Hints have costs: a hinted copy is not readable on its owner until replayed, so a quorum read can miss it; a stand-in may die with its hints; and hint storage loads a cluster that is already a node short. Decide per keyspace whether a stand-in’s copy counts towards W: yes for carts (Dynamo’s sloppy quorum, and what “always writable” means), no for device bindings. Cassandra’s ANY level is the extreme: a stored hint alone satisfies the write.

Repair on read. When replicas disagree, the coordinator writes the newest version back, as in the quorum figure; Dynamo’s coordinator even lingers after replying to fix late, stale replicas. It is cheap and aimed at the keys people use, but cold keys are never fixed.

Repair in the background: Merkle-tree anti-entropy. For everything else — hints lost with a stand-in, cold keys, a disk that lost a file — replicas compare data. Each builds a Merkle tree per partition: leaves hash small key ranges, parents hash their children. Matching roots mean the partition is in sync; otherwise the replicas descend only into branches that differ.

Each node holds about 1.25 billion keys in about 102 partition replicas, roughly 12.2 million keys each. A depth-15 tree has 2¹⁵ = 32,768 leaves of about 370 keys. If one key differs, the replicas swap about two hashes per level — 30 hashes — and then one leaf’s 370 keys, instead of 12.2 million key hashes. The cost is disk: building a tree reads every key. Cassandra runs repair on an operator schedule, in incremental form (data written since the last repair) and full form (which also catches corruption), and recommends repairing every node at least every 7 days with its 10-day grace period. Dynamo throttled all background work — repair, handoff, streaming — with an admission controller that watched foreground latency, such as how close the last minute’s p99 disk read latency was to 50 ms.

How does a single node store data fast?

At peak each node handles 900k ÷ 120 = 7,500 replica writes and 2M ÷ 120 ≈ 16,700 replica reads a second over 2.9 TB and 1.25 billion keys. (Background on both engine families: reads and writes: how databases store and fetch data.)

Bad: an append-only log with an in-memory hash index. Writes are purely sequential and a read is one disk access, but the index must hold every key: 1.25 billion × about 70 bytes (key, offset, overhead) ≈ 87 GB of RAM per node, rebuilt by scanning the log after each restart.

Good: a B-tree with a write-ahead log. Sorted pages updated in place, upper levels cached: a point read is about one disk read, with very predictable latency. DynamoDB’s storage replicas use a B-tree plus a write-ahead log, and most Dynamo instances used Berkeley DB. The cost is on writes: changing a 2 KB value rewrites a whole 4–16 KB page at a random location, plus a log record.

Great for this write-heavy mix: a log-structured merge (LSM) tree, the design Bigtable described and Cassandra and RocksDB use. It turns random writes into sequential ones and pays for it later, in the background.

The write path, matching the figure:

1. Append to the commit log, a sequential write. This is what survives a crash (see write-ahead logging).
2. Insert into the memtable, a sorted structure in memory, and acknowledge the coordinator.
3. Flush. At about 1 GB, freeze the memtable, start a new one, and write the frozen one to disk as an immutable sorted file — an SSTable — in level 0; its commit-log segment can then go. At 15 MB/s per node, that happens about every 67 seconds.
4. Compact. Background merges combine SSTables, drop overwritten versions and keep the number of files a read must check small.

Reads check the memtable, then SSTables from newest to oldest, and return the first version found. Each SSTable keeps a Bloom filter in memory; at about 10 bits per key it gives about 1% false positives, so skipping four files wastes 4 × 0.01 = 0.04 disk reads on average. An in-memory block index then points at the one block to read — Bigtable described a lookup as a single disk seek — and missing keys usually touch no disk at all.

How durable is “appended to the commit log”? Cassandra’s default periodic mode acknowledges at once and fsyncs every 10 seconds, so a crash can lose up to 10 seconds; batch mode fsyncs before acknowledging. Losing an acknowledged write under periodic sync needs both zones’ replicas to crash in the same window — rare, but a power event is exactly that kind of correlated failure. Use group commit, as Bigtable did: gather writes for about 2 ms and fsync once. At 7,500 writes/s that batches 7,500 × 0.002 = 15 writes into each of 500 fsyncs a second, easy for NVMe.

Compaction is where the bill arrives. Each byte is rewritten several times on its way down the levels — write amplification — as RocksDB’s tuning guide works through for leveled compaction.

For our node: 15 MB/s × about 33 ≈ 500 MB/s of compaction writes at peak — sustainable on NVMe, but it eats read I/O and wears flash. So choose per keyspace. Leveled compaction keeps space amplification near 1.1 and reads to one file per level: use it for read-heavy profiles and device bindings. Tiered compaction (RocksDB’s universal) merges whole runs of similar size, cutting write amplification at the cost of more files per read and, the guide warns, up to double the space during compaction: use it for write-heavy sessions and carts. When most data has a TTL, group files by time window so whole files expire without rewriting. And budget compaction against foreground latency, as Dynamo did for all background work.

Deletes are writes: tombstones and the grace period. Suppose two replicas erase a key while the third is down. When it returns, repair sees “one has a value, two have nothing” and copies the value back — a zombie. So a delete writes a tombstone, a timestamped marker that travels through hints and repair and shadows older versions; expired TTL values become tombstones too. After a grace period — Cassandra’s gc_grace_seconds, 10 days by default — compaction may drop a tombstone, but only in a compaction that also includes every older SSTable holding data it shadows. Hence the rule above: repair every replica inside the grace period. Keys deleted and rewritten constantly pile up tombstones that reads must step over; model them as fresh keys with TTLs instead.

How do we add or remove nodes without downtime?

Membership changes are explicit — an operator or automation decides, never the failure detector. The goal: move about 2.9 TB per added node during a sale without quorums ever ceasing to overlap.

Bad: pause and reshuffle — that is downtime — or random tokens with scan-based streaming, which made Dynamo’s bootstrapping take almost a day in peak season.

Good: virtual nodes with range streaming, which is how Cassandra bootstraps: the joining node streams each new range from a current replica and serves when done. It works, but ranges are random slices that must be found inside SSTables, and many move at once.

Great: move fixed partitions as files, under a plan. A cluster manager computes the target layout and drives it through the partition map, one step at a time.

1. Join as a pending owner. The manager picks about 102 partition replicas for node 121, taking them from the most loaded donors, and publishes map v813 marking node 121 as a pending replica for each. It never moves two replicas of the same partition at once: Cassandra’s documentation warns that simultaneous range movements can break consistency and even lose data.
2. Stream and dual-write. From v813, coordinators also send writes for those partitions to node 121, not yet counted towards W. Each donor sends a snapshot of the partition’s SSTables — immutable files, so a snapshot is a set of hard links and the transfer a sequential copy. At about 400 MB/s, 2.85 TB takes 2.85 × 10¹² ÷ (4 × 10⁸) ≈ 7,100 s, about 2 hours, throttled like any background task.
3. Flip. When a partition is complete, the next map version makes node 121 a full replica and drops the donor. A client with a stale map is redirected by the node it contacts; Dynamo’s clients polled for membership every 10 seconds and refreshed at once on hitting an unreachable node.
4. Clean up. Donors delete their copies after a safety delay, once the new replica has passed a repair.

Removing a node is the reverse, with the leaving node as donor. Replacing a dead node streams from surviving replicas, plus a repair if it was gone longer than the hint window. Bigger machines simply get more partitions.

The final design

Put the pieces together. Every storage node runs the same software; the only special components sit off the request path.

Follow one cart update and the background work around it:

1. Route. The client library hashes carts/user:98231 and finds its replicas in a cached partition map, refreshed every 10 seconds and on redirects: B in zone a, D in zone b, C in zone c.
2. Coordinate. It sends the put to B, a replica in its own zone.
3. Write. B appends to its commit log and memtable, forwards to D and C, and acknowledges once B and D have it (W = 2).
4. Read. A QUORUM get reads B’s copy plus a digest from D, with a speculative request to C if D is slow; stale replicas are repaired before the reply.
5. Hint. If C is down, B — or a stand-in — keeps a hint and replays it, throttled, when C returns.
6. Detect. Every node gossips once a second and runs a phi-accrual detector.
7. Repair. Anti-entropy compares Merkle trees for every partition at least weekly, inside the 10-day grace period, on a background I/O budget.
8. Grow. The cluster manager publishes a new partition map; node 121 takes dual writes, streams partition files, then serves reads.

Check it against the requirements. Latency: one local LSM operation plus one round trip to the faster of two other zones — assuming a millisecond or two between zones, p50 is about 1–2 ms, and speculative requests keep a slow replica from setting the p99, which stays under 10 ms. Throughput: 2.9M replica operations/s ÷ 120 ≈ 24k per node, 60% of the budget; about 36k with a zone down. Availability: any two replicas, or stand-ins, accept a write, so a node or a zone can fail. Durability: acknowledged writes sit in group-committed logs in two zones, and hints and repair restore the third copy. Elasticity: a new node serves fully in about two hours.

What interviewers look for by level

Follow-up questions

What if one key becomes extremely hot?

A ₹99 phone deal goes live and its stock key gets 200k reads a second. However you partition, one key lives on three machines, and reading at ONE still gives each 200k ÷ 3 ≈ 67k — far above the 40k budget. Absorb reads before they arrive: an in-app cache with a 100–500 ms TTL turns them into a few hundred a second, fine for a stock display (hot keys). Split a hot counter into 16 sub-keys on different partitions and sum on read, or use a CRDT counter. Splitting partitions won’t help: DynamoDB notes that a partition whose traffic goes to a single item gains nothing from a split.

What if we need two regions, say Mumbai and Hyderabad?

Keep three replicas per region, six in all, and use region-aware levels. Cassandra’s LOCAL_QUORUM waits for a majority in the coordinator’s region, so latency stays local and clients homed there read their own writes; remote copies are written without waiting. EACH_QUORUM waits for a majority in every region, adding a cross-region round trip and failing whenever a region is cut off. Use LOCAL_QUORUM for carts and sessions, with the same conflict machinery for keys written in both regions, and route writes that need one global winner to a home region. More in multi-region and active-active.

What if a disk silently corrupts data?

Replication copies corruption as faithfully as it copies data, so detect it at every boundary. Store a checksum with every block and verify it on read; checksum every message between nodes and every streamed file. DynamoDB goes further: it checksums every log entry, message and log file, and continuously scrubs replicas, checking that all three match and that they match a replica rebuilt from archived logs. Here, full Merkle repair plays a similar role: a replica whose hashes disagree is repaired from the other two.

What if we need range scans, such as all of one user’s orders?

Hashing scatters adjacent keys across the cluster, so a scan would touch every node. The usual fix is a composite key: hash only a partition key (the user) to place data, and keep items within it sorted by a sort key (the order time) — DynamoDB’s model, where the partition key goes through a hash function and the sort key orders items with the same partition key. Fully ordered range partitioning, as in Bigtable, makes any scan cheap but invites hot spots on sequential keys, and needs ranges that split and merge.

What if a user asks us to erase their data for good?

A delete only writes a tombstone; the old value still sits in older SSTables until compaction runs after the grace period, and in backups. To promise erasure within a deadline, give that keyspace a shorter grace period (which means repairing it more often), trigger compaction of the affected partitions once the grace period passes, and expire backups on a matching schedule. Bigtable’s authors made the same point about major compactions: they reclaim space from deleted data and make sure it disappears in a timely fashion, which matters for sensitive data.

How to say it in the interview

“I’ll design a leaderless, Dynamo-style key-value store for carts, sessions and similar data: 100 TB, a million reads and 300 thousand writes a second at peak, p99 under 10 ms, and writes that are never refused because a node or a zone is down.

Keys hash onto a ring split into 4,096 fixed partitions, each on three nodes in three zones, and a small etcd-style store holds the partition map. A partition-aware client sends each request to a replica, which coordinates it. Writes go to all three; with N = 3 and W = R = 2, read and write sets overlap, and callers can choose ONE, QUORUM or ALL per request. Quorums give overlap, not linearizability, so device binding uses compare-and-set over per-partition consensus.

Concurrent writes are detected with version vectors. Carts keep siblings or use an observed-remove-set CRDT, so neither adds nor removes are lost; sessions use last-write-wins, where losing one of two simultaneous writes is harmless.

Failures are handled in layers: gossip every second with a phi-accrual detector; hinted handoff with a sloppy quorum; read repair; and weekly Merkle-tree anti-entropy, inside the 10-day tombstone grace period, so deletes never come back.

Each node runs an LSM tree — group-committed commit log, memtable, SSTables with Bloom filters — with compaction chosen per keyspace, because leveled compaction rewrites each byte about 33 times. That sizes to 120 nodes, 40 per zone, about 2.9 TB each, enough to survive losing a zone. Adding a node streams about 2.9 TB of partition files from many donors with dual writes, in roughly two hours and with no downtime.”

Concepts used in this design

• Consistent hashing — the ring, virtual nodes and how little data moves when membership changes.
• Database sharding — partitioning data across machines and choosing the partition key.
• Quorums and leaderless replication — N, R and W, sloppy quorums and what overlap does and does not promise.
• Data replication — leaderless versus leader-based replication and their trade-offs.
• CAP theorem and PACELC — why an always-writable store gives up consistency during partitions and latency the rest of the time.
• Strong vs eventual consistency — picking the guarantee per keyspace and per request.
• Clocks, ordering and conflict resolution — version vectors, last-write-wins and CRDTs.
• Consensus and leader election — the compare-and-set path for keys that need one winner.
• Gossip protocol — spreading membership and heartbeats without a central registry.
• Heartbeats — the signal behind phi-accrual failure detection.
• Write-ahead logging — the commit log that makes acknowledged writes survive a crash.
• Reads and writes: how databases store and fetch data — B-trees versus LSM trees.
• Bloom filters — skipping SSTables that cannot contain a key.
• Cassandra and DynamoDB — the production descendants of this design.
• Hot keys, cache stampede and invalidation — protecting a single very popular key.
• Multi-region and active-active — extending the store across regions.

References and further reading

1. Dynamo: Amazon’s Highly Available Key-value Store (DeCandia et al., SOSP 2007)allthingsdistributed.comThe original leaderless design: virtual nodes, preference lists, (N, R, W) = (3, 2, 2), vector clocks and the shopping-cart merge, sloppy quorums, Merkle trees, gossip, and the three partitioning strategies with production numbers.
2. Amazon DynamoDB: A Scalable, Predictably Performant, and Fully Managed NoSQL Database Service (Elhemali et al., USENIX ATC 2022)usenix.orgHow the managed successor differs: Multi-Paxos per partition with a leased leader, B-tree plus write-ahead-log replicas, gray-failure detection, why splits cannot fix a single hot item, partition and sort keys, checksums and continuous verification.
3. Apache Cassandra documentation: Dynamo architecturecassandra.apache.orgToken ring and vnodes and their downsides, last-write-wins timestamps, consistency levels (ONE, QUORUM, LOCAL_QUORUM, EACH_QUORUM, ANY), gossip every second, and the phi accrual failure detector.
4. Apache Cassandra documentation: Storage enginecassandra.apache.orgCommit log, memtables, immutable SSTables and their Bloom filters, and periodic (10 s) versus batch commit-log sync.
5. Apache Cassandra documentation: Hintscassandra.apache.orgHinted handoff timeline, the 3-hour hint window, idempotent replay, and the worked example of how long a hint backlog takes to replay under the default throttle.
6. Apache Cassandra documentation: Tombstonescassandra.apache.orgWhy deletes are written as tombstones, zombie data, gc_grace_seconds (10 days by default), and when compaction may drop a tombstone.
7. Apache Cassandra documentation: Repaircassandra.apache.orgMerkle-tree anti-entropy, full versus incremental repair, and repairing every node at least every 7 days to stay inside the grace period.
8. Apache Pekko documentation: Phi Accrual Failure Detectorpekko.apache.orgA clear description of Hayashibara et al.’s detector: phi = −log10(1 − F(t)), why jitter changes detection time, and a default threshold of 8.
9. RocksDB Tuning Guide (RocksDB wiki)github.comWrite, read and space amplification; the leveled-compaction example (write amplification ≈ 33, space ≈ 1.14); universal compaction; 10 Bloom bits per key ≈ 1% false positives.
10. Bigtable: A Distributed Storage System for Structured Data (Chang et al., OSDI 2006)static.googleusercontent.comWhere memtables, SSTables, minor and major compactions, deletion markers, group commit and per-SSTable Bloom filters were first described together.