Design a Distributed Key-Value Store: One Node to a Cluster

A key-value store maps an opaque key to an opaque value and answers one question very fast: what is stored under this key? The hard part is not the lookup. It is deciding which machine owns the key, making the write survive crashes, keeping copies in agreement, and saying precisely what a reader may see while things are failing. This page builds one design, step by step, from a map in a single process to a partitioned, replicated, region-aware database, and each component appears only after the failure that demands it.

The running example is fictional. Meridian Profiles is a B2B platform that stores user preferences, account settings and tenant configuration as durable records for three tenants: Atlas (large and busy), Birch and Cedar. Every size, rate and latency on this page is a teaching assumption, not a measurement, and no figure describes a real product. The records we follow are atlas/user42/settings, birch/user17/settings and one hot configuration key, atlas/feature=checkout_policy. One tracked mutation, evt-024, is a Put of Atlas user 42's settings; the Visual Design Journey embedded further down follows it from request to commit, through a crash, and into later conflicts.

What “key-value” does and does not mean here

A key is a short byte string that identifies a record; a value is an opaque blob (bytes or JSON) that the store does not interpret. Because the store never looks inside values, it can spread records across machines by key alone. That is the whole bargain: you give up joins, secondary indexes and multi-key transactions, and in exchange a single-key lookup touches one partition and scales by adding machines.

It is also a *database*, not a cache. A cache may drop an entry whenever it likes because the source of truth lives elsewhere. Meridian's settings records are the source of truth: a successful Put that later disappears is data loss, not a miss. That one sentence rules out silent eviction, forces a durability boundary, and is why the Redis-style "fast in-memory map" is a starting point we will have to leave, not a destination. (The Redis fundamentals lesson covers that engine on its own terms.)

Point lookups versus range-heavy workloads

Meridian reads atlas/user42/settings by its exact key. A workload that instead says "give me every event for user 42 between two timestamps" is range access, and it wants keys stored in sorted order on one partition. Ordered storage and hash distribution pull in opposite directions; chapter 7 returns to that choice. For now, note that point lookup is the operation everything below optimizes, and that a range scan is a feature the partitioning model must explicitly permit rather than something every key-value store gets for free.

Visual Design Journey scenes 1 and 2 show this starting point: one caller, one settings write, and then the list of requirements that the rest of the story has to satisfy.

Why large values and secondary indexes stay out

Both exclusions are decisions about this workload, not limits of the idea. Large values. Meridian's records average about 1 KiB and are capped at 256 KiB because every byte of a value is paid for several times: it is written to the log on three replicas, rewritten by compaction, and shipped again in snapshots and region replication, and one very large value can delay the writes queued behind it in the replicated log. The usual alternative is to keep the blob in object storage and store a small record here holding its location, size and checksum. The store's durability then covers the pointer, and the application owns the order (upload first, then write the pointer) and the cleanup of orphaned objects.

Secondary indexes. Finding every user with locale = de-DE needs a second ordering over the same data. Keeping it consistent with the primary record means a write touches more than one partition, or a read fans out to all of them. Meridian reads by exact key, so it keeps the write path simple and feeds a separate index or search system from the change stream, accepting that this index is eventually consistent and can lag. If attribute lookups were on the hot user-facing path, this is the first exclusion to revisit (Trade-offs & Decisions, Decision 17).

Requirements for a data store are mostly promises about failure, so they have to be written as promises. "Fast and reliable" decides nothing; "an acknowledged write survives the loss of any one machine, and a read in the strong mode never returns anything older than the last acknowledged write" can be tested.

Functional behavior

Put(key, value, expectedVersion?, idempotencyKey) stores a value and returns a committedVersion that the server assigns; the same number is exposed as an ETag. With expectedVersion the write is conditional: if the key's current version differs, the server answers 412 Precondition Failed and changes nothing. Get(key, readMode) returns the value and its version. Delete(key, expectedVersion?) is a versioned write that records a deletion marker. Every record therefore has a version that rises with each committed change, and that version is what lets two writers avoid overwriting each other by accident.

Non-functional targets

The targets below are the ones the design must meet; the numbers are assumptions chosen so that the arithmetic in the next chapter is checkable.

Choosing the consistency contract on purpose

Linearizability means that every operation on a single key appears to take effect at one instant between its start and its finish, and that those instants respect real-time order: once a write has been acknowledged, every later read sees it or something newer. It is a statement about single objects. Serializability is about multi-object transactions and is a different property; the V1 store offers the first for one key and none of the second. Meridian chooses linearizable single-key semantics in the home region because settings are the kind of data where a user saves a preference and immediately reloads the page, and an old value on that reload reads as a bug.

The cost is stated up front. When a partition cannot assemble a majority, the strong mode refuses writes and strong reads with 503 rather than guessing. This is what the CAP framing actually says: not "pick any two of three" in general, but *during a network partition*, an operation either stays consistent and may be rejected, or stays available and may be inconsistent. A leaderless design makes the opposite choice and is shown as a clearly labeled alternative in chapter 10, never blended into the reference.

The response vocabulary is fixed here so that later chapters can lean on it. 200/201 carries committedVersion. 404 means the key does not exist or was deleted (the product may choose to expose deleted and expired separately). 409 is a conflict such as reuse of an idempotency key with a different payload. 412 is a failed version precondition. 421 or a redirect, or a retryable WRONG_EPOCH, tells a client its routing map is stale. 429 is per-tenant throttling. 503/UNAVAILABLE means no quorum or no safe leader. And a client *timeout* is neither success nor failure: it is an unknown outcome, retried only with the same idempotency key.

Visual Design Journey scene 2 shows the same requirement list as it is built up one line at a time.

Capacity arithmetic here has one job: decide whether the next component is justified. For Meridian's assumed workload the answer is that one machine cannot hold or serve it, and that the binding constraint is throughput and hot keys, not raw disk.

The inputs, all assumed: 2,000,000,000 records averaging 1 KiB (1,024 bytes, key plus value plus metadata, before indexes, replicas or write amplification). Average load is 200,000 reads/s and 40,000 writes/s; short peaks reach 1,000,000 reads/s and 200,000 writes/s.

A planning worksheet, clearly assumed

Split the key space into 1,024 logical partitions: 2.048 TB / 1,024 = 2.0 GB each, about 1,953,125 records. With three replicas that is 3,072 partition replicas; spread over 96 nodes it is 32 replicas per node and 6.144 TB / 96 = 64 GB of replicated data per node. Assume a node can apply 20,000 replica writes per second *or* serve 25,000 point reads per second, and that mixed load adds as fractions.

At the peak, 200,000 writes/s become 600,000 replica applies (each write lands on three replicas), or 6,250 per node: 31.25% of the write budget. One million reads over 96 nodes is about 10,417 per node: 41.67% of the read budget. Together that is about 72.9% of a node, for a short burst. Counted as node-equivalents, 600,000 / 20,000 + 1,000,000 / 25,000 = 30 + 40 = 70; dividing by a 70% utilization target gives 100 nodes ideal against 96 planned. That is a planning illustration for a brief peak, not a promise, and nothing here is an average-utilization claim.

Disk is the small number

Replicated data of 6.144 TB plus an assumed 5% for indexes, Bloom filters and metadata is 6.4512 TB. If volumes are kept at most 60% full (assumed, to leave room for compaction scratch space, tombstones, snapshots and repair), the minimum provisioned capacity is 6.4512 / 0.6 = 10.752 TB, about 112 GB per node across 96 nodes. That is tiny next to the CPU and I/O the fleet needs, so this workload is throughput-bound, not capacity-bound. WAL space, backups and snapshots are additional line items to list explicitly; folding them into one magic "2×" would hide the reasoning.

Requests per second also understate physical work. If, for illustration, one client write causes roughly ten bytes of disk write for each byte accepted once log, flush and compaction rewrites are counted (an assumed factor of 10×; real values depend on the compaction strategy and workload and must be measured), then the peak is 204.8 MB/s × 3 replicas × 10 = 6,144 MB/s, about 6.1 GB/s cluster-wide or 64 MB/s per node. The average gives 1,228.8 MB/s, or 12.8 MB/s per node. One logical write is not one disk I/O.

Skew beats averages

Dividing traffic by node count assumes uniform spread. Real traffic is not uniform. In a simplified 100-partition illustration (distinct from the 1,024-partition fleet above), average reads of 200,000/s give 2,000 per partition. If atlas/feature=checkout_policy alone takes 60,000 reads/s it is 30× the average partition and 30% of average fleet reads. With 25,000 point reads per second per replica, 60,000 / 25,000 = 2.4, so at least three serving replicas are needed, and three replicas leave only 80% utilization. Chapter 13 handles that key; here it is a design pressure, because no amount of adding nodes lowers the load on one key's leader.

Visual Design Journey scene 3 puts these numbers on screen.

The first version is the one anyone would write: a dictionary in the application process. It is instant, it is trivially correct for one thread, and it loses every record the moment the process ends.

Everything about it is attractive for ten thousand records: a lookup is a hash probe, a write is an assignment, and there is nothing to configure. Now trace evt-024. Atlas user 42 saves new settings; put stores them in memory and returns version 8 to the caller. Ten seconds later the process is restarted by a deployment, or the machine reboots, or the kernel kills it for memory. The next get returns nothing, and the user's saved preferences are simply gone, even though the API said success.

This is the first invariant breaking, so name it precisely. The system acknowledged a write at a point where only volatile memory held it. The durability boundary is the point at which a write will survive the failure mode you promised to survive. For V0 that boundary is "until this process exits", which is below the requirement. Nothing in this design is wrong in logic; it fails because the promise and the mechanism do not match.

A natural first patch is to periodically dump the whole map to a file. That narrows the loss window to the dump interval but does not remove it, and writing a multi-gigabyte file while accepting writes either blocks them or produces an inconsistent snapshot. We need a mechanism that makes each write durable individually and cheaply. Visual Design Journey scenes 4 and 5 show the map answering instantly and then forgetting evt-024 on a crash.

A write-ahead log (WAL) is an append-only file where every change is recorded before it is applied anywhere else. If the process dies, restart reads the log from the start and rebuilds memory. The acknowledgement moves from "I updated a variable" to "the log has it on stable storage".

Appending is the cheapest write a disk offers: sequential, no seeking, no rewriting of existing pages. Each record carries a sequence number, an operation, the key and value, and a checksum so recovery can tell a complete record from a torn one. The order of operations is the whole trick: append the record, make it durable, apply it to the in-memory map, then reply. Reversing the first two steps recreates V0's lie.

Append is not durable; fsync is

Calling write() hands bytes to the operating system's page cache, which is memory. A power loss at that moment loses them. fsync (or an equivalent flush such as fdatasync or direct I/O with flush semantics) asks the OS to push the data to the storage device and wait until the device confirms. Only after that returns can the record be called durable on stable storage, and even then the claim holds *unless the storage stack honours flushes*: a drive or controller with a volatile write cache that ignores flush requests can still lie. Production fleets verify this with power-loss testing rather than trusting a datasheet.

fsync is also expensive, so doing one per write would cap throughput at the device's flush rate. Group commit fixes this: many writers' records are appended and a single fsync covers the whole batch. Using the worksheet figures, a node applying 6,250 writes/s with an (assumed) 1 ms commit window collects about 6.25 records per fsync. The price is that each write waits up to the window before its acknowledgement; the gain is that the flush cost is shared.

Recovery and checkpoints

On restart the engine opens the log, verifies each record's checksum, and applies records in order. A crash can leave the final record half-written; the checksum fails, recovery stops at the last good record and truncates the tail, and because that record was never acknowledged (the ack follows the flush), nothing the client was told is lost. Replaying a month of writes would be slow, so the engine periodically writes a checkpoint (a consistent snapshot of the map tagged with the last sequence it includes) and discards log segments older than the checkpoint.

Replay must be idempotent or position-aware, since a crash during recovery means replaying again; recording each record's sequence number and skipping those already applied is the usual approach. With this in place evt-024 survives the restart that erased it in V0. Visual Design Journey scenes 6 and 7 show the append-then-flush ordering and the replay after a crash.

With a WAL, durability is solved but the working set still lives in memory. At 2 TB of logical data it cannot. The single-node engine now needs a layout on disk that keeps writes sequential and makes reads findable without scanning everything. The common answer is the LSM tree (log-structured merge tree).

The idea has four parts. Writes go to the WAL and then into a memtable, a sorted in-memory structure. When the memtable reaches a size threshold it is written out as an SSTable (sorted string table), an immutable file whose keys are in order, and the covered WAL segment can be dropped. Reads look in the memtable first and then in SSTables from newest to oldest. Because files are never edited in place, an update or delete is simply a newer entry that shadows older ones. Compaction merges several SSTables into fewer larger ones, keeping only the newest version of each key and reclaiming the space of shadowed entries.

A key read through several files

Follow atlas/user42/settings over a week. Version 6 was flushed into SSTable 1 on Monday. A write on Tuesday made version 7, which flushed into SSTable 3. On Wednesday version 8 (evt-024) arrived and is still in the memtable. A Get finds version 8 in the memtable immediately. After the memtable flushes, a Get must check SSTable 4 (newest), find version 8 and stop. But a read for birch/user17/settings that was last written long ago might have to look at every file before finding it, and a read for a key that does not exist has to prove its absence in *every* file.

That proof of absence is what Bloom filters are for. A Bloom filter is a compact bit array and a set of hash functions: adding a key sets k bits, and a lookup says "definitely not here" if any of its bits is zero or "maybe here" otherwise. A correctly built filter has no false negatives, so a "no" is a safe skip. With 10 bits per key and k = 7 hash functions the false-positive rate is (1 − e^(−7/10))^7 ≈ 0.82%. For one partition replica of 1,953,125 keys that is 19,531,250 bits, about 2.44 MB, and across 32 replicas per node about 78 MB per node. An absent key probing five SSTables expects about 5 × 0.0082 = 0.041 needless disk reads. A sparse index block inside each SSTable then narrows a hit to a single data block.

What compaction costs

Compaction is where the design pays for cheap writes. Every byte may be rewritten several times as it moves down through levels or merges with others; this is write amplification, and with read amplification (files consulted per read) and space amplification (stale versions kept) it forms a three-way trade-off. Different strategies pick different corners: size-tiered compaction writes less but leaves more overlapping files and more space overhead; leveled compaction keeps fewer files per read at the price of more rewriting; time-window strategies suit data that expires together. Compaction also competes with foreground traffic for disk and CPU, so it has to be rate-limited and observed, an operational matter chapter 13 revisits.

Visual Design Journey scenes 8 and 9 show writes accumulating into SSTables and the read path getting longer as files pile up. For the formats themselves, the architecture deep dive gives a WAL record layout, an SSTable sketch and the read algorithm in pseudocode.

One engine on one machine eventually runs out: disk, memory for filters, or simply the throughput of its storage. Meridian's peak of 600,000 replica writes and a million reads per second against 20,000 and 25,000 per node budgets means one node is nowhere close. The data must be partitioned: divided into pieces that different nodes own.

The tempting shortcut: hash modulo node count

Hash each key and take hash mod N. It is simple and spreads uniformly. Then add a fourth node to a cluster of three. A key keeps its owner only when h mod 3 == h mod 4, which is true for 3 of every 12 residues: 25% stay and 75% move. In bytes, 0.75 × 2.048 TB = 1.536 TB of logical data would have to be copied (4.608 TB counting three replicas), during which routing is wrong for much of the key space. Growth that should be routine becomes an outage-grade reshuffle.

Virtual partitions: decouple keys from machines

Instead of mapping keys to nodes directly, map keys to a fixed, larger number of logical partitions, then map partitions to nodes in a table. The fixed key-to-partition rule never changes (fixing the partition count up front is what makes that possible); only the partition-to-node table does. Meridian's worksheet uses 1,024 partitions. The journey scene uses a smaller picture, 12 virtual partitions: with three nodes each owns four. Add a fourth node and move three partitions (one from each of the old nodes), so each node owns three. That is 25% moved, all of it onto the new node, and 0.512 TB of the logical 2.048 TB. Not zero movement, and not magic, but proportional to the capacity added rather than to everything. Consistent hashing achieves a similar bound with a ring and virtual nodes; the fixed-partition table is the variant that makes a partition a first-class unit you can replicate, move and monitor.

Hash versus range placement

Hash placement spreads load evenly but destroys key order, so a prefix scan would have to ask every partition. Range partitioning keeps adjacent keys together, which makes Scan(prefix) cheap and tablet-style splits natural, but sequential or popular key ranges can concentrate on one partition, and the split points must be managed. Meridian's reference uses hash partitioning for point lookups with a tenant-scoped composite key, and offers Scan only on collections it deliberately places on range-partitioned data. Neither choice is universally right: it depends on whether the dominant access is a point lookup or a range.

Visual Design Journey scenes 10 to 13 show the single node filling, the modulo reshuffle, and the same cluster growing with virtual partitions.

Once data lives on many nodes, every request has to find the right one, and the answer changes whenever the cluster does. The partition map records, for each partition, its current owners and an epoch: a number that rises each time ownership changes.

A map entry reads something like partition 417 → leader node-12, replicas {node-12, node-31, node-77}, epoch 9. A client library or a stateless router hashes the key (after prefixing the tenant), finds the partition, looks up the owner and sends the request along with the epoch it believes is current. The map lives in a small strongly consistent control store, and routers cache it. Caching is what makes routing fast and also what makes it dangerous.

Stale routers

Suppose partition 417 moves from node-12 to node-90 and the epoch goes from 9 to 10. A router that has not refreshed still sends evt-024 to node-12 tagged epoch 9. If node-12 simply accepted it, two nodes could believe they own the partition and writes would diverge. Instead node-12 knows it has handed ownership over, sees the stale epoch and answers 421 or a retryable WRONG_EPOCH with the new map version. The router refreshes and retries. Invariant: a key resolves to one partition and routing epoch, and a stale route produces a redirect, retry or bounded rejection, never a second authoritative owner.

A safe cutover sequence

Moving a partition is a sequence of small steps because each step has to be crash-safe. First the target node is added as a non-voting learner and copies a snapshot while the source keeps serving. Second, it catches up on log entries written since the snapshot. Third, when the lag is small, the control store atomically changes the map to the new owner with epoch 10. Fourth, the source stops accepting writes for that partition, forwards stragglers, and eventually deletes its copy after a safety delay. If anything fails before step three, the source remains authoritative and the move is abandoned; if the control store is unreachable, no cutover happens. Movement is throttled, because copying a 2.0 GB partition competes with foreground work, and the migration's progress is a dashboard item so a stuck move is noticed.

Visual Design Journey scenes 14 to 16 walk through the partition map, a stale router being redirected and the rebalance with epoch cutover.

Partitioning raised capacity and lowered durability: lose the one node that owns partition 417 and its 2.0 GB, and everything that was acknowledged for those keys, is unavailable or gone. The fix is replication, keeping the partition on several nodes. But replication alone is only copying; what makes it safe is a rule for when a write counts as committed.

Meridian uses a replication factor of three (RF=3) with the replicas in separate failure domains (different racks or availability zones), so one rack or power feed does not take them all. For partition 417 the three replicas are named A, B and C. A is the leader: it orders every write for the partition by appending it to the replicated log; B and C are followers that receive the log from the leader. Each partition elects its own leader, so leadership is spread across nodes and the whole cluster is not funnelled through one machine.

What “committed” means

A quorum is a majority of the voting replicas: with three replicas, two. Leader A writes evt-024 to its own WAL (log index 118, term 4), sends it to B and C, and counts itself plus any follower that confirms a durable write. When B confirms, two of three hold it; the leader advances its committed index to 118, applies the entry to its state, and replies to the client with version 8. C may be slow, and that is fine: the commit did not wait for it. No follower commits an entry before the leader has seen a majority; the arrows in the diagram go outward and the commit decision comes back to the leader.

Why a majority? Any two majorities of three overlap in at least one replica. If a new leader must win votes from a majority, at least one voter has every committed entry, and the election rule only favours candidates whose logs are at least as up to date. That is how a new leader is guaranteed to hold everything a previous leader acknowledged. Raft formalises this; the design here is Raft-like as a conceptual model and does not claim to be any specific product's internals.

Replication also has a price: three replicas triple write work (the 600,000 replica applies at peak in chapter 3) and add a network round trip, within the home region, to every commit. Visual Design Journey scenes 17 to 19 show a single copy being lost, the replica group appearing, and evt-024 committing on a majority.

With replicas comes a question that did not exist on one node: when a client reads, which copy answers, and how old can its answer be? The store exposes that choice as a named read mode, because the right answer differs by call site.

Take evt-024. Leader A commits version 8; the client gets success. Follower C has not yet applied index 118. A follower-stale read of atlas/user42/settings routed to C returns version 7, which is a legal answer for that mode and a bug if the product promised something stronger. A linearizable read is served by A, which first confirms it is still the leader (either by exchanging heartbeats with a majority to establish a read index, or under a time-bounded lease), and returns 8. For a caller that just saved a setting, read-your-writes gives the better cost: the client keeps the committedVersion as a floor, and any replica that has not reached it either waits briefly or hands the request on.

Stale reads are a feature when you ask for them

Serving every read from the leader would pin 1,000,000 reads/s onto 1,024 leaders. Follower reads let read-heavy, staleness-tolerant calls (dashboard views, background jobs) use the other replicas. The contract is that the *API* says which mode was used, so a client never receives stale data while believing it is fresh.

When the quorum is gone

A partition that loses its majority has to choose. The reference store is CP in the CAP sense for strong operations: the minority side refuses writes, and refuses linearizable reads, answering 503. The reason is concrete. If the minority side kept accepting writes, and the majority side also elected a leader and accepted writes, two histories would exist for the same key and no automatic rule could pick one without discarding acknowledged writes.

A leaderless design makes the opposite choice, and the journey shows it as a labeled alternative in scenes 27 to 30. In that model a coordinator sends each write to N replicas and waits for W acknowledgements, and a read asks R replicas and takes the newest. With N=3, R=2 and W=2, R + W = 4 > 3, so every read quorum overlaps every write quorum in at least one replica. That overlap is an intuition for how a read can find a recent write, not a linearizability guarantee: sloppy quorums that use stand-in nodes, concurrent writes, clock skew, partially failed writes and membership changes all break it. Such stores answer conflicts with version vectors or siblings, or by last-writer-wins on timestamps, and they repair divergence with hinted handoff, read repair and anti-entropy. Cassandra, for instance, treats hinted handoff as best-effort and relies on anti-entropy repair for the eventual-consistency guarantee. It is a valid design for a different contract; the reference does not borrow its guarantees.

Visual Design Journey scenes 20 to 23 show a stale follower read, the linearizable read, the minority partition rejecting a write, and leader fencing; scenes 27 to 30 are the labeled leaderless branch, and scene 30 returns to the reference.

Two admins open Atlas user 42's settings at the same time. Both read version 7. One changes the theme, the other changes the notification policy, and both save. Without a guard, the second save silently overwrites the first, and nobody ever sees an error. This is the lost update.

The guard is compare-and-set (CAS), exposed as Put(key, value, expectedVersion). The write applies only if the key's current version equals expectedVersion; otherwise it is rejected with 412 and the caller re-reads, merges and retries. What makes this safe is *where* the comparison runs. If the client reads version 7, checks it, and then issues an unconditional write, another client can sneak in between the check and the write. A check-then-put in application code is not CAS. The comparison and the update must be one atomic step inside the replicated state machine, executed by the partition leader in log order.

Walk it through. Client 1 sends Put(atlas/user42/settings, ifVersion=7) and client 2 sends the same a moment later. Leader A appends them to the log as entries 118 and 119. Both entries contain the precondition, and the state machine applies entries in order on every replica. When entry 118 applies, the version is 7, so it succeeds and the version becomes 8. When entry 119 applies, the version is 8, the precondition fails, and the state machine records the outcome as a rejection. Every replica reaches the same conclusion because they apply the same entries in the same order, and the leader returns 412 to client 2, with the current version so it can retry without another read. Exactly one writer wins.

The same idea for deletes and for retries

Delete(key, expectedVersion) uses the same mechanism. Retries are a separate hazard: a client that times out does not know whether its write committed. It retries with the same idempotency key, a client-supplied identifier stored with the committed entry for a documented retention window. The state machine recognises the repeated key and returns the original result instead of applying twice; reusing a key with a different payload is a 409. See the idempotency lesson for the general pattern, and race conditions for how this failure presents in production.

The limits matter too: CAS protects one key. Updating two keys together atomically needs transactions, which V1 does not offer. Visual Design Journey scenes 24 and 25 show the two-writers race and the ambiguous timeout.

Replication tolerates a dead follower trivially. A dead leader is harder because everyone has to agree on a successor, and a leader that is only *believed* dead must be stopped from continuing.

Each partition group counts terms: a term is a number that increases every time a leader election starts. Followers expect heartbeats from the leader. If a follower hears nothing for a randomized election timeout, it increments the term, becomes a candidate and requests votes. The Raft paper suggests randomized timeouts in the range of 150–300 ms (implementations differ), the randomness being there so that candidates rarely tie. A candidate wins when a majority votes for it, and a voter grants its vote only if the candidate's log is at least as up to date as its own. That rule is what carries committed entries across leaders.

The old leader that does not know

Say leader A in term 4 stalls for a garbage-collection or I/O pause. B and C elect B as leader in term 5. A wakes up still believing it leads. If A accepted evt-025, it could send it to C, but C is in term 5 and rejects messages from term 4, and A cannot gather a majority anyway. Because every request, vote and log append carries the term, a stale leader discovers the truth the first time it talks to anyone and steps down. This is fencing: using a monotonically increasing token (term, or epoch for routing) so that actions by an obsolete actor are refused. For reads, a leader that served a linearizable read under a lease must stop doing so when its lease expires, which is why leases depend on bounded clock drift.

Recovery from disk

A crashed replica restarts, loads its latest snapshot, replays its durable log, and asks the leader to fill in what it missed. If its log has entries the new leader never committed (written by A in term 4 but not replicated), the leader's log wins and those uncommitted entries are truncated. They were never acknowledged, so no client was promised them. If the replica has been gone so long that the leader has already discarded the log it needs, it receives a fresh snapshot instead. Recovery traffic is throttled so a returning node does not starve foreground work.

Visual Design Journey scenes 22, 23 and 26 show the minority side, leader fencing, and a replica recovering from its log; the failure side of this chapter is the subject of the failure and scale scenarios page.

A store that is correct at rest can still collapse under its own housekeeping or under one popular key. Growth adds three kinds of pressure the earlier chapters did not need to name: background work that competes with foreground traffic, skew that defeats even spreading, and one tenant's burst landing on everyone.

Compaction debt

Chapter 6 showed that compaction pays for cheap writes. If writes arrive faster than compaction can merge them, SSTables accumulate; every read must check more files, read latency rises, disk fills with stale versions, and eventually the engine has to slow or stall writes to let compaction catch up. The accumulation is compaction debt. Speeding compaction up helps the debt but steals disk and CPU from foreground reads, so the cure is a *bounded* rate: enough to keep debt flat, with a visible P99 latency impact. The metrics worth watching are pending compaction bytes, files per read, and free disk against the 60% target from chapter 3.

The hot key

Return to atlas/feature=checkout_policy, a configuration value every Atlas checkout reads. In the simplified 100-partition illustration, average partition load is 2,000 reads/s and this single key draws 60,000/s: 30× the average partition. A replica serves 25,000 point reads/s under the assumed budget, so 60,000 needs at least 3 serving replicas, and three leave 80% utilization and little headroom. Moving the partition to a bigger node, or adding more partitions, does nothing for this key, because it still lives in exactly one partition.

The options all trade something. Letting followers serve the key spreads reads over three replicas but returns follower-stale data, which is acceptable for a policy that changes rarely if the product says so. A client or near-cache with a short TTL removes most reads from the store altogether, at the price of staleness up to the TTL; the cache is optional and explicitly layered on top, not a change to the store's semantics. Request coalescing merges simultaneous identical reads so one backend read serves many callers. Splitting the key into copies is valid only for read-mostly data. And writes to that key remain serialized through one leader: salting or fanning out a key does not help a single hot key's writes, because a conditional update must still be ordered in one log.

Per-tenant budgets

Atlas's backfill job starts pushing 150,000 writes/s. Without limits it eats the shared commit pipeline and Birch's writes see queueing. The store needs per-tenant quotas enforced at the front (requests, bytes, concurrent operations), a bulkhead so that overflow in one tenant's queue cannot occupy every worker, and a visible 429 telling Atlas to slow down, instead of latency creep for everyone. See bulkheads, rate limiting and load shedding. Rebalancing and repair are throttled for the same reason, and during overload the store follows an explicit policy: reject new work with 429 or 503 before accepting writes it cannot finish.

Visual Design Journey scenes 34 to 36 show compaction pressure, the hot key, and the hot-tenant isolation.

Deleting a record sounds like the easiest operation: remove it. In a replicated, immutable-file store, removing the bytes from one place does not make the record gone, and may let it come back.

Atlas deletes atlas/user42/settings at version 9. If the leader simply erased its copy, a replica that was offline and still holds version 8 could, when it returns, look like the only copy and teach everyone that version 8 exists. Immutable SSTables add a second reason: an older file still contains the old value, and the engine cannot edit it. So a delete is written as a new, versioned, replicated entry: a tombstone that says "as of version 9, this key is deleted".

Reads treat a tombstone as 404 while it is the newest version; older values beneath it are shadowed. Tombstones are themselves data and are kept for a tombstone retention period (Cassandra calls it gc_grace_seconds; its documented default is 864,000 seconds, or 10 days) and then dropped during compaction. The rule that sets the length: retention must exceed the longest time a replica can be out of the cluster or unrepaired before it is repaired. If a replica returns after 11 days against a 10-day retention, the tombstone may already be gone and the replica's old value can come back as a zombie. This page uses 10 days as a teaching value.

Repair keeps replicas honest

Lagging replicas converge by replaying the log in the reference design; in addition a background anti-entropy process compares replica contents (typically by comparing hash trees of key ranges) and fixes differences, bounded in rate so it does not compete with foreground traffic. Read repair, where a read notices stale copies and updates them, belongs to leaderless designs; in the leader-based reference the log is the repair path and the leader's committed order is authoritative. Whichever mechanism is used, repairs must carry tombstones along with values, or repair itself would spread the zombie.

TTL is a delete scheduled by a clock

A record can carry an expiry time. This is not cache eviction: eviction drops an entry for memory pressure and is allowed to forget, whereas expiry is a promise that the record is absent after a given time and present before it. Because it depends on time, clock skew matters: a replica whose clock runs fast may hide a record early. The reference decides expiry from a timestamp agreed through the log (the expiry is part of the entry and checked against a bounded-skew clock on read), and at physical removal time writes a tombstone through the normal path so replicas agree.

The consequence is stated in the contract rather than avoided. A deleted or expired value is logically absent immediately, while old copies persist in older SSTables until compaction rewrites them, in snapshots until they age out and in backups until retention ends (an example policy of 35 days means up to 35 days after a delete before the bytes leave backups). Erasure is bounded by those processes, never instant, and any privacy-driven deletion promise has to be worded to match. Visual Design Journey scenes 31 to 33 show the logical delete, the resurrection hazard and TTL expiry, and scene 37 covers backup and restore.

Customers in the US and EU want local latency, and a whole region can fail. Stretching a quorum across oceans makes every commit pay a wide-area round trip, so the reference keeps the commit path inside a home region and ships the log onward asynchronously.

Meridian's records are homed in EU: the partition leader and its follower quorum live there, and commits finish in-region. A US replica receives the committed log asynchronously. Reads in the US served locally are eventual: they can be behind. If a US caller needs the latest value, the request is routed to the EU home region and pays the round trip. Nothing about asynchronous replication gives global strong consistency; it gives locality for reads and a warm copy for recovery.

Lag becomes data-loss exposure

Suppose the US replica trails by 300 ms (assumed). EU has acknowledged version 11 of a record; US shows version 10. If the EU region disappears at this moment, whatever has not yet arrived is lost to a failover. With EU receiving 30% of average writes (an assumed 12,000 writes/s), up to 12,000 writes/s × 0.3 s = 3,600 acknowledged-but-unreplicated writes are at risk at that instant. That figure is the RPO (recovery point objective: how much acknowledged data can be lost) in concrete terms. It depends on lag, and it is never zero. The RTO (recovery time objective: how long to restore service) is the other number; any value you state is a labeled example.

Failover, done safely

A safe regional failover needs, in order: a decision that EU is really gone (a human or a quorum of observers outside both regions, to avoid split brain); a fencing step so a returning EU cannot keep accepting writes in the old epoch; promotion of US to home with a new epoch; an explicit statement that remote-written data may be missing the last RPO window and that readers of linearizable see the new region's history; and after EU returns, reconciliation of the unreplicated tail rather than silent merge. Residency matters too: records bound to EU must not be silently promoted into a US region, so failover for them follows the residency policy, which may mean staying unavailable.

Visual Design Journey scenes 38 to 42 follow evt-024 through the home region, the lag, the outage, the safe failover and a labeled multi-active alternative.

Every part of the finished design exists because an earlier version failed. Clients reach a router that consults a partition map; each partition is a replicated group with a leader and two followers, each replica running a WAL and an LSM engine; background compaction, repair and rebalancing run under rate limits; a control plane owns the partition map and epochs; quotas protect tenants from one another; and a home region commits in-region and replicates to remote regions asynchronously.

One Put, one Get, one CAS, two outages

A Put for evt-024: the router hashes the tenant-scoped key to partition 417, reads the cached map at epoch 9, and sends the call to leader A; A checks quota and the idempotency key, appends to its WAL (index 118, term 4), replicates to B and C, commits when two of three have flushed, advances its committed index, applies to the engine and returns version 8. A Get in linearizable mode is confirmed by the leader and returns version 8; in follower-stale mode it may return version 7 from C. A conditional write with ifVersion=8 succeeds once and the other racing writer gets 412. A single-node outage costs one replica's headroom or a brief election; a region outage promotes US with the stated RPO. Visual Design Journey scenes 44 to 46 replay the put path, the get and failure path, and the final mental model.

What it guarantees, and what it does not