Key-Value Store Architecture: Sharding, Replication & LSM Trees

A key-value API is small on purpose, and every verb in it has to say what happens when the network, a replica or the client misbehaves. The reference store exposes Put, Get, Delete and optional BatchGet, and each response either names a committed version or says plainly that the outcome is unknown.

The running example is fictional: Meridian Profiles stores user preferences and tenant configuration for tenants Atlas, Birch and Cedar. Every size and rate on this page is a teaching assumption, not a measurement — 2 billion records, an average record of 1 KiB, a key limit of 1 KiB and a value limit of 256 KiB. Keys are always tenant-scoped composites such as atlas/user42/settings, and the tracked mutation evt-024 is a Put to exactly that key.

The four calls

Put(tenant, key, value, ifVersion?, idempotencyKey) writes an opaque value (bytes or JSON) and returns a committedVersion the server assigned, also exposed as an ETag. Get(tenant, key, readMode) returns the value plus its version. Delete(tenant, key, ifVersion?) is a write like any other: it creates a *tombstone* version rather than erasing anything on the spot. BatchGet fans independent point reads across partitions and returns per-key results; it is a convenience, not a transaction, so two keys in one batch may reflect different moments. A limited Scan(prefix/range) exists only when the partitioning scheme keeps neighbouring keys together (the partitioning chapter explains why hashing gives it up). V1 has no joins, no cross-key transactions, no secondary indexes and no full-text search — each of those is a different product.

Reading the status codes

404 means the key has no live value. Whether the API tells a deleted or expired key apart from one that never existed is a product choice; exposing it helps debugging and sync clients, but it also reveals history, so say which you picked. 412 Precondition Failed is reserved for a compare-and-set mismatch: the caller supplied ifVersion and the committed version differs. 409 Conflict is a different situation — reusing an idempotency key with a different payload, or hitting a key range that is mid-migration. A stale routing view produces a redirect (421 or 307 style) or a retryable WRONG_EPOCH. If no majority of replicas can commit, the answer is 503 / UNAVAILABLE, never a cheerful success. 429 is per-tenant throttling. The point of splitting these is that clients react differently: 412 means re-read and merge, 409 means fix the request, 503 and 429 mean back off.

Timeouts are not failures

Suppose Atlas's client sends evt-024, the leader commits it, and the acknowledgement is lost on the way back. The client sees a timeout. If it concludes 'failed' and sends the same update again without protection, a non-idempotent mutation (a counter increment, an append) applies twice. The fix is an idempotency key: the server records (tenant, idempotencyKey) → result *inside the same replicated log entry* as the write, so a retry after a leader change still finds it. The record is kept for a documented window — say 24 hours, an assumed policy — after which a reused key is treated as new. A retried conditional write is also safe from double application, because the second attempt carries the now-stale ifVersion and gets 412. On its own, though, a 412 cannot tell the client whether it was its own first attempt or a competing writer that produced the new version; the idempotency key is what lets the server answer with the original result instead.

Values, limits and versions

The version is a monotonically increasing per-key number assigned when the entry is applied in log order, so it is identical on every replica. It is not a wall-clock timestamp, which is what makes CAS comparisons immune to clock skew. Size limits (key ≤ 1 KiB, value ≤ 256 KiB in this lesson) exist because a single oversized record occupies a whole replication round trip, inflates compaction and can block a leader's apply loop; Google documents guidance of roughly 10 MB maximum per Bigtable value, but a settings store has no business approaching that.

Each replica needs a local engine that makes writes durable quickly and reads findable cheaply. The widely used answer is a log-structured merge (LSM) tree: append every change to a log, keep recent data sorted in memory, and periodically write sorted immutable files that are merged in the background.

The write-ahead log

A write-ahead log (WAL) is an append-only file where a change is recorded *before* it is applied anywhere else. Appending is sequential, which storage handles well, and it gives crash recovery something to replay. Each record carries enough to verify itself:

The write call write() only copies bytes into the operating system's page cache. Durability on stable storage requires fsync (or an equivalent flush), and even that assumes the storage stack honours flushes — drives or controllers with volatile write caches can still lose an acknowledged flush unless the stack is configured correctly. So the durability boundary has to be stated: *acknowledged after the record is flushed on this replica* (and, in the replicated design, committed by a majority).

fsync is expensive relative to memory, so engines use group commit: concurrent writers queue behind one flush. With the assumed figure of 6,250 write-applies per second on a node and a 1 ms commit window, one flush covers about 6.25 records on average — the cost per record falls as load rises, at the price of up to a millisecond of added latency. That is a latency knob, and it is a durability knob too: flushing less often is faster and widens the loss window.

Recovery

After a crash, startup reads the WAL from the last checkpoint, verifies each record's checksum, replays valid records into a fresh memtable, and truncates at the first bad or incomplete record. Records the client was never acknowledged for may or may not survive; acknowledged ones must. A concrete timeline: evt-024 was appended and flushed at seq=88213, acknowledged, and the process died before the memtable was flushed to an SSTable. On restart the replay reaches 88213, rebuilds the value for atlas/user42/settings, and a Get returns v8.

Memtable, SSTables and compaction

New writes land in a mutable in-memory sorted structure, the memtable (a skip list or balanced tree). When it reaches a size threshold it is frozen and written out as an SSTable — a sorted, immutable file of key-value entries divided into blocks with an index, optionally a Bloom filter, and a checksum footer. After the flush completes, the WAL segment covering it can be recycled.

Since SSTables are never modified, updates and deletes simply add newer entries (keeping several versions of one key side by side is the core of multi-version concurrency control, MVCC: readers pick the version they are entitled to see, and writers never edit a version in place); older ones are shadowed until compaction merges runs and discards what is no longer visible. Strategies trade differently: size-tiered merges runs of similar size (lower write amplification, more overlap for readers to check), leveled compaction keeps non-overlapping sorted runs per level (fewer files to check per read, more rewriting), and time-window compaction groups data by age for time-ordered workloads. Cassandra documents all three; which one is right depends on workload and must be measured.

Reading through many files

A point read checks the memtable, then SSTables from newest to oldest, stopping at the first entry for the key. Naively that is one disk read per file. A Bloom filter is a compact bit array that can say 'definitely not in this file' or 'possibly in this file'. With 10 bits per key and 7 hash functions the false-positive rate is about 0.82% (classic math: (1 − e^(−7/10))^7 ≈ 0.0082), and a correctly built filter has no false negatives. For one partition replica holding 1,953,125 keys, that is 19,531,250 bits ≈ 2.44 MB; across 32 replicas on a node, about 78 MB of filters. A lookup for an absent key probing 5 SSTables expects only about 5 × 0.0082 ≈ 0.041 needless disk reads.

Filters answer membership for *point* lookups; range scans cannot use them, which is one reason scans over many overlapping files are costlier. Read amplification is the number of places consulted per read, write amplification is physical bytes written per logical byte (WAL, flush and every compaction rewrite — this lesson uses an assumed illustrative 10×), and space amplification is disk used over live data, high while old versions and tombstones await compaction.

Once the data no longer fits on one node's disks or throughput budget, the key space is divided into partitions (shards), each owned by a set of nodes. The routing rule must give every key exactly one owner at a time and must survive adding capacity without reshuffling everything.

Why `hash mod N` fails

Hash the key, take it modulo the node count: simple and uniform. Then grow from 3 to 4 nodes. A key stays put only when h mod 3 == h mod 4, which holds for 3 of every 12 residues — 25% stay and 75% move. For the 2.048 TB logical data set that is 1.536 TB to relocate (4.608 TB counting three copies), during which routers and nodes disagree about owners. Consistent hashing or, more simply, a fixed number of virtual partitions fixes this: keys map to a stable partition id, and only the *partition-to-node* assignment changes. Adding a fourth node to three should move ≈25% (0.512 TB logical) — all of it onto the new node — and a scene illustrates this with 12 virtual partitions: 4 per node becomes 3 each, with 3 partitions moving. Movement is reduced, not zero.

Meridian's worksheet (assumed) uses 1,024 logical partitions: 2.048 TB / 1,024 = 2.0 GB each, about 1,953,125 records. With RF=3 that is 3,072 partition replicas, and on 96 nodes 32 replicas per node. Partitions are small enough to copy in minutes-to-hours and numerous enough to balance.

Two classic ways to get that stable mapping are worth naming. Consistent hashing places nodes (and keys) on a hash ring and gives each key to the next node clockwise, so adding a node steals only the arc next to it. Giving every physical node many small arcs, called virtual nodes, smooths out uneven arcs and lets a bigger machine own more of them; when a node dies its arcs scatter across many survivors instead of landing on one neighbor. Range-based stores go the other way and divide the sorted key space into tablets that split and merge as they grow. Meridian's fixed 1,024 partitions are effectively virtual nodes with an explicit map: simple to reason about, with the map held in the control plane rather than derived from the ring.

Hash versus range

Hash partitioning applies a hash to tenantId/key (hashing the whole composite so a tenant prefix cannot be forged into another namespace and so tenants spread out). Distribution is even, point operations are cheap, and a Scan would have to ask every partition. Range partitioning (Bigtable-style tablets, or ordered stores such as FoundationDB, described only as an ordered transactional key-value store) keeps keys sorted so prefix scans touch few partitions, but monotonic keys such as timestamps pile onto the last range and splits must be managed. Meridian's primary workload is point lookup, so it hashes; a limited Scan is promised only on a range-partitioned deployment.

The partition map and epochs

A partition map (kept in a small, strongly consistent control-plane store, which could be a Raft-replicated group such as etcd) records who owns each partition. Every entry carries an epoch that increments on any ownership change:

Routers cache the map. A request carries the epoch the router believes; the node compares it with its own. If they differ, the node refuses with WRONG_EPOCH plus the new epoch, and the router refreshes and retries. That is how stale routing becomes a redirect instead of two nodes both accepting writes for the same key.

Moving a partition without losing writes

Rebalancing is a state machine: the control plane marks the partition MOVING at epoch *e*, the destination installs a snapshot and then tails the source's log until it is caught up, checksums of both sides are compared, writes to the source are paused for a brief fence while the last entries drain, the map flips to the destination at epoch *e+1*, and only then does the source stop serving and later delete its copy. A failure before the flip rolls back by discarding the destination copy; after the flip the epoch makes the source refuse. Throttling the copy keeps foreground P99 latency visible and bounded.

Two practical details sit around that state machine. Before the flip, the destination can serve *shadow reads*: the router sends a small sample of reads to both owners, discards the destination's answer, and compares the two, so a bad copy shows up before any client depends on it. After the flip, routers learn the new epoch lazily, so some requests still reach the old owner for a while; those are refused with WRONG_EPOCH and retried, which is why the old copy is kept for a grace period rather than deleted at once. When many partitions have to move (a new node joining), the control plane staggers them, a few at a time under a bandwidth cap, rather than shifting all traffic in one step; rollback is a pre-flip abort that discards the half-built copy, and a post-flip problem is handled by moving the partition back as a new, ordinary migration at epoch *e+2*.

Skew and the hot tenant

Averages hide concentration. With a simplified 100-partition illustration, 200,000 reads/s averages 2,000 per partition; one hot key such as atlas/feature=checkout_policy at 60,000 reads/s is 30× that average and 30% of average fleet reads. Adding nodes does not redistribute a single key. Chapter 10 treats mitigation, and tenant-level isolation appears in the multi-tenancy chapter.

Copying a partition to three machines does not by itself make it safe. What makes a replicated key-value partition correct is an agreement rule: a write counts as committed only once a majority of voting replicas have durably stored it in the same log order, and only a replica holding every committed entry can become leader.

The reference design uses a Raft-like model per partition. 'Raft-like' is deliberate: this is a conceptual reference for explaining the mechanics, drawing on the Raft consensus algorithm, not a claim that every product behaves this way, and not a promise that a replication factor of 3 alone yields consensus. Three copies written by an ad-hoc script with no term, no log matching and no election rule is just three copies that can disagree. etcd, covered in a live lesson on this site, is a production example of a Raft-based store.

The log and its vocabulary

A term is a monotonically increasing election number; at most one leader exists per term. logIndex orders entries; committedIndex is the highest index known to be stored on a majority; the state machine (the key-value map) applies entries in log order up to committedIndex. The entry's version is derived from the apply order, so replicas agree on it.

A write through three replicas

Replica A is leader, B a follower, C a follower that is lagging. evt-024 arrives at A. A does a cheap pre-check of the idempotency table and the ifVersion precondition to reject obvious failures early (the authoritative check repeats at apply time, as the conditional-write section shows), appends the entry at index 88213 in term 9 and flushes its own WAL. It sends AppendEntries to B and C. B appends, flushes and replies; C is slow. As soon as A has the entry durable on itself and B — two of three, a majority — it advances committedIndex, applies the entry (producing v8) and replies to the client. C catches up later from the log. The client never waited for C, and no follower committed anything before the leader said it was committed.

Raft requires that a leader never overwrites its own log and that a follower rejects an AppendEntries whose preceding entry (index and term) it does not hold; that log-matching check lets a returning replica truncate its divergent tail and adopt the leader's. A new leader also may not count replicas of an old-term entry as committed on its own; it commits one entry from its current term, which indirectly commits everything before. These are the details that make 'copy to a majority' actually safe.

Elections, fencing and stale leaders

Followers expect heartbeats. If one hears nothing for a randomized election timeout (the Raft paper that introduced the algorithm suggests values such as 150–300 ms; implementations differ), it increments its term and asks for votes. A replica grants at most one vote per term, and only to a candidate whose log is at least as up to date as its own. That second rule is what guarantees that any winner holds every committed entry: a committed entry sits on a majority, any winning majority overlaps it, and a voter with the entry will not vote for a candidate lacking it.

Now the dangerous case. A is partitioned from B and C but does not know it. B and C elect B in term 10. When the partition heals, A still believes it leads term 9. Any message from A carries term 9; B and C reject it because their term is higher, and A, on seeing the higher term, steps down. This is fencing: the term is a token that makes an old leader's requests harmless. For side effects outside the replicated log, a fencing token must also be checked by the resource itself.

Conditional writes in the replicated state machine

Two clients read v7 of atlas/user42/settings and both send ifVersion=7. Both requests are ordered by the leader's log. The first is applied: version becomes 8, response 200. The second is applied against version 8, the precondition fails and it returns 412 with the current version — and importantly the check is evaluated deterministically at apply time on every replica, so the outcome is the same everywhere.

Membership changes, snapshots and catch-up

Changing the replica set (replacing a dead node, moving a partition) is itself dangerous: if old and new configurations could each form an independent majority, there would be two leaders. Raft handles this with joint consensus (decisions need a majority of both old and new configurations) or with one-server-at-a-time changes. A new replica is brought up by installing a snapshot — a point-in-time copy of the state machine at some logIndex — and then replaying the log after it; the leader can compact its log up to the snapshot point, which bounds disk but means a replica that was down longer than the retained log must take the snapshot path.

Some stores skip the leader. Any replica among the N that hold a key can coordinate a request; writes go to W replicas, reads ask R, and conflicts are reconciled afterwards. It is a different contract, and everything below is labeled ALTERNATIVE because its guarantees must not be mixed into the reference design.

The moving parts

A coordinator (any node, often the one the client contacted) finds the key's preference list — the N replicas responsible, derived from consistent hashing — and sends the operation to them. With no central map, nodes usually learn who is alive and who owns which arcs by gossip: each node periodically swaps its membership view with a few random peers, so news spreads through the cluster in a handful of rounds, though two nodes can disagree about membership for a short time. With N=3, R=2, W=2 a write succeeds after two acknowledgements and a read waits for two replies. R+W = 4 > N = 3, so every read set intersects every write set in at least one replica. That is quorum-overlap intuition: the read will contact at least one replica that saw the latest *completed* write.

Repair mechanisms

Hinted handoff lets a stand-in node hold a write for an unreachable replica and replay it later; Cassandra treats it as best-effort: hints do not replace anti-entropy repair as a guarantee of eventual consistency, and hints are only kept for a limited window (recent default configs show three hours — check your version). Read repair compares replies on a read and writes the newest back to stale replicas. Anti-entropy uses Merkle trees (hash trees over key ranges) to find divergent ranges between replicas cheaply and stream the difference. Repair is scheduled work with real I/O cost, and it has to run more often than the tombstone retention window, as the lifecycle chapter explains.

Merkle trees: finding the one range that differs

Comparing two replicas key by key would mean shipping every key. A Merkle tree avoids that. Each replica splits a key range into leaf ranges and hashes the content of each (every key with its version, deletions included), then hashes pairs of hashes upward to one root. Take four leaf ranges A to D. Replica R1 has leaf hashes 3f, 9c, 71, e2; replica R2 has 3f, 9c, 5a, e2 (illustrative values, not real hashes). The roots differ, so they compare children: the A+B side matches and is skipped entirely, while the C+D side differs. Comparing C and D shows that only C differs. Only the keys in range C are exchanged.

Inside range C the two replicas swap (key, version) pairs. R1 holds atlas/user42/settings at v7 and R2 holds v6, because R2 was down when v7 was written. The newer version wins and is streamed to R2. A delete must be compared as a version too: if tombstones were ignored, a deleted key would look like a missing key and be copied back to life. The cost is that building a tree means reading the data, and every write changes a leaf hash, so trees are built from snapshots or maintained incrementally, and repair is scheduled work rather than something a request waits for.

Keep gossip and Merkle repair apart. Gossip spreads small facts between nodes: who is alive, who owns which range. Merkle comparison finds differences in the data itself. A node can be marked alive by gossip and still hold stale data that only repair will fix. In the leader-based reference, membership and ownership come from the epoch-numbered partition map and from terms and heartbeats, not from gossip, and a follower that fell behind catches up from the leader's ordered log or a snapshot. The reference still uses a Merkle diff in repairReplica (see the pseudocode chapter), but for a narrower job: confirming that replicas that have been away for less than the tombstone retention window agree again. In a leaderless design there is no single log to replay, so anti-entropy is the main way replicas converge.

Concurrent writes: siblings or last-write-wins

Two clients update atlas/user42/settings on different coordinators at the same moment. Without a leader there is no log order. Version vectors attach a counter per writer-node so replicas can tell whether one version descends from another or the two are concurrent; concurrent ones are kept as siblings and returned to the application to merge (the Dynamo paper's shopping cart is the classic example). Last-write-wins instead keeps the value with the larger timestamp: simple, but it silently discards one of the writes, and with clock skew can discard the later one. Bigtable's documentation describes server-time last-write-wins for conflicting same-cell writes across clusters in multi-cluster routing.

Neither is better in the abstract. A shopping-cart-like record that can be merged and must always accept writes suits the alternative; tenant settings guarded by CAS suit the reference. DynamoDB sits partly outside this framing: its documentation describes eventually consistent reads by default with optional strongly consistent reads within a Region, and global tables with multi-Region modes — check the vendor's current documentation rather than relying on summaries.

A read mode is a promise about *how stale* an answer may be. The reference store offers four, named the same everywhere: linearizable, follower-stale, read-your-writes and eventual. Linearizability means every operation on a key appears to take effect at a single instant between its start and its response, consistent with real time; it concerns one object and is not the same as serializability, which concerns multi-object transactions.

Why the leader alone is not enough

Suppose A was leader in term 9 and has been silently replaced by B in term 10. A client still talking to A reads A's local state and gets a value missing a write B committed. Answering from the leader without checking it is still leader is not linearizable. Two correct techniques: a read-index round (the leader notes its current commit index, confirms with a majority heartbeat that it is still leader, waits to apply up to that index, then answers), or a leader lease (the leader assumes it stays leader for a bounded time that followers promise not to undermine; correctness then depends on bounded clock drift). Both cost something: a round trip or a clock assumption.

Follower reads, honestly labeled

A follower that serves its local state may be behind. In the example, C is lagging at v7 while A has committed v8; follower-stale read from C returns v7. That is acceptable when labeled — the response carries the applied index, optionally a staleness bound such as 'at most 500 ms behind' — and unacceptable when presented as current. Follower reads spread load, for instance for the 60,000 reads/s hot key, but only for callers who opted into staleness.

Read-your-writes gives a client its own writes without full linearizability: after evt-024 commits at v8 the client holds a session token (the version, or log index, at which its write committed) and sends it as a floor. A follower serves the read only if it has applied at least that point; otherwise it waits briefly or forwards to the leader. Eventual reads (a remote region, or the leaderless alternative) promise only convergence if updates stop. Causal consistency, which preserves cause-before-effect across keys, sits between these but needs dependency tracking and is outside V1.

Deleting from a system that keeps immutable files and several replicas is a write, not an erase. The delete has to be replicated and remembered long enough that no stale copy can bring the value back.

Tombstones

A tombstone is a versioned marker meaning 'this key was deleted at version v'. It goes through the same log as any write, so every replica learns of it; reads that find a newer tombstone than any value return 404. The old value still sits in older SSTables on disk until compaction merges the files and sees that the tombstone shadows it. The value is *logically* absent immediately and *physically* present for some time.

Zombie resurrection

Suppose replica C is down when atlas/user42/settings is deleted. A and B hold the tombstone. They compact and, having kept the tombstone past its retention window, purge it. Later C returns carrying the old value. Repair sees C has a value and the others have nothing, and copies it back: the deleted record is alive again. The rule to prevent it: tombstone retention must exceed the longest time a replica can be out of the cluster or unrepaired before it is repaired or replaced. A replica gone longer than that must be rebuilt, not allowed to rejoin with old data. Cassandra's documented gc_grace_seconds default is 864,000 seconds (10 days); this lesson uses 10 days as its assumed teaching policy.

In the leader-based reference, followers apply tombstones in log order and a lagging follower catches up from the log or a snapshot containing them, which narrows the window; but restores from backup, a long-isolated replica or an operator copying files reintroduce the same risk, so the retention rule still applies.

TTL is not cache eviction

A time-to-live attaches an expiry to a record, and Meridian's TTL is a durable, replicated rule: a cache may evict whenever it likes, this store may not drop an unexpired record. Expiry is typically evaluated at read time (a record past its deadline is treated as deleted) and materialised later by a background sweep that writes a real tombstone through the log, so all replicas agree. Because 'now' differs between machines, clock skew can make a record vanish early on one node, so the leader stamps the expiry deadline into the entry, and reads compare against a bounded-skew clock.

Physical erasure and backups

Privacy requests need honesty about timing. After a delete commits, the live value is gone from reads. Compaction later removes it from the replicas' files — when that happens depends on strategy and load. Snapshots and backups hold older copies until they expire: with an assumed 35-day backup retention, physical removal from backups could take up to 35 days. State this window; never promise instant erasure. Redaction of specific records from backups, or per-tenant encryption keys that can be destroyed (crypto-shredding), are options with their own cost.

Deleting a whole tenant is harder than deleting a key, because hashing scatters a tenant's records over every partition. A workable sequence: first mark the tenant deleted in the control plane (and bump a per-tenant generation number that is part of every stored key), so reads and writes for it fail immediately and old-generation entries are logically invisible; then run a throttled background sweep that writes tombstones for the tenant's keys partition by partition, so the work is visible and cannot starve foreground traffic; then let compaction remove the bytes. If the tenant's data is encrypted under its own data key, destroying that key (crypto-shredding) makes the remaining copies, including those in backups, unreadable well before the 35-day backup window ends. The tenant should be told which of these have completed and which are still pending.

Restore needs a consistency point: a snapshot per partition is paired with log entries up to a chosen index or timestamp. Anything deleted after that point must be replayed again, or removed through a deletion ledger, or the restore itself resurrects deleted data.

Many tenants share the same partitions, WAL and compaction threads, so isolation has to be built into the key, the request path and the background work, not added as a filter at the end.

The tenant identity comes from a verified credential (a signed token or mutual-TLS identity) and never from a field the caller types. The server prepends it to the key, so user42/settings from Atlas becomes atlas/user42/settings, and every later layer — routing hash, storage key, log line, admin API — sees the composite. A caller cannot ask for birch/… because the namespace is applied after authentication, not parsed from input. Key length (≤ 1 KiB) and value size (≤ 256 KiB) bounds apply before anything reaches the log.

Noisy neighbors

Per-tenant quotas (requests, bytes written, scan work) are enforced at the edge with a token bucket (each tenant's bucket refills at its allowed rate and every request spends tokens; an empty bucket means rejection) and answered with 429; the rate limiting pattern explains the mechanism. Inside the cluster, bulkheads (separate queues or thread pools per tenant tier) prevent one tenant's burst from filling shared queues. If Atlas's traffic dominates a partition, the hot tenant can be given dedicated partitions or nodes, moved with the same safe migration path.

Encryption, keys and operators

Data is encrypted in transit (TLS between clients, nodes and regions) and at rest. Envelope encryption lets a key-management service hold a master key and wrap per-tenant or per-volume data keys, so rotation re-wraps small keys instead of rewriting terabytes. Administrator access is a separate, audited path with least privilege and break-glass approval; operator actions such as moving a partition or reading raw data are logged with who, what and when.

• Derive the tenant from the verified credential and apply it to namespace, routing, storage, logs and admin operations.
• Redact keys and values from traces and metrics; log tenant id and key hash instead.
• Enforce size limits and quotas before the write path.
• Audit every administrative read and mutation.

• Accepting tenant ids from request bodies.
• Putting raw values or full keys in error messages and metric labels.
• Using tenant id as an unbounded metric label (cardinality explosion).

Spreading a key-value store across regions trades latency, consistency and availability explicitly. The reference design keeps a home region (EU) as the authoritative location for each partition's leader and quorum, and ships the log asynchronously to a remote replica (US).

What asynchronous replication gives

Writes commit within EU at EU latency. The US replica applies the log with a lag; in the running example it is 300 ms behind. A US eventual read may return v10 when EU has acknowledged v11, and must be labeled so. Callers who need the latest value route to the EU leader and pay the cross-ocean round trip. Asynchronous replication does not make the whole system globally strongly consistent, because the remote copy is allowed to be behind.

Region loss and RPO

If EU fails abruptly, writes acknowledged in EU but not yet shipped are lost to the US copy. Assume EU takes 30% of the 40,000/s average writes — 12,000 per second — and lag is 0.3 s: up to 12,000 × 0.3 = 3,600 acknowledged writes are exposed at the instant of failure. That number is RPO exposure (recovery point objective), and it depends on lag, never on a promise of zero. Recovery time (RTO) is how long until service resumes. Both are shown here as illustrative; a real contract names them with measured lag.

Promoting the US replica requires fencing the old EU leader so it cannot accept writes if it revives (a region-level epoch or a control-plane decision recorded in a strongly consistent store), accepting that v11 may be missing in the promoted copy (it holds v10), and deciding what to do with the lost tail when EU returns. Records with data-residency rules must not be silently promoted in the US; their failover has to stay in a compliant region or fail closed.

Synchronous alternatives and vendors

Putting voting replicas in several regions gives stronger guarantees across regions at the price of cross-region latency on every commit, and a region-wide outage must leave a majority. Multi-active designs let several regions accept writes and reconcile conflicts, so they are a different model from a globally coordinated strongly consistent product. Bigtable's documentation describes replication across clusters as eventually consistent by default, stronger behavior only with single-cluster routing, and reverting to eventual consistency after a failover until replication catches up. Cassandra's LOCAL_QUORUM keeps quorum in the local datacentre for latency. DynamoDB global tables are multi-active; AWS describes two modes, multi-Region eventual consistency (MREC) and a separately configured multi-Region strong consistency (MRSC) — read the vendor's current documentation for exact behavior and supported operations, since details change. Earlier DynamoDB global tables behavior was that transactions were atomic only in the originating Region, so partial results could be visible elsewhere during replication.

The fleet is sized by throughput and amplification, not by raw data volume. Everything here is a worked planning illustration with assumed budgets, not a measurement.

Footprint

2,000,000,000 records × 1,024 B = 2.048 TB logical (≈ 1.86 TiB). Three copies give 6.144 TB of raw record replicas — which is not provisioned disk. Adding an assumed 5% for indexes, Bloom filters and metadata gives 6.4512 TB; keeping volumes at most 60% full (an assumed target leaving room for compaction temporary files, tombstones, snapshots and repair) gives 6.4512 / 0.6 = 10.752 TB minimum, about 112 GB per node across 96 nodes. WAL space, backups and snapshots are listed separately, not hidden inside a 2× factor.

Throughput budget

The workload is throughput-bound: 112 GB of provisioned disk per node is small next to the CPU and I/O needed. Little's Law (L = λ × W) sizes concurrency: 1,000,000 reads/s × an assumed 5 ms is 5,000 reads in flight, and 200,000 writes/s × an assumed 20 ms commit latency is 4,000 writes in flight — roughly 52 and 42 per node.

Network, CPU and catch-up time

Each committed write crosses the network more than once. At the assumed peak of 200,000 writes/s × 1 KiB, the leaders send every entry to two followers, so replication traffic is 2 × 204.8 = 409.6 MB/s across the cluster, or about 4.3 MB/s per node over 96 nodes; the acknowledgements are tiny by comparison. The asynchronous remote copy ships only the home region's share: at the assumed 12,000 writes/s that is about 12.3 MB/s, roughly 98 Mbit/s, before any compression. CPU goes to checksums (CRC32C is commonly hardware-accelerated, so the cost is usually modest, but measure it), serialization, TLS and compaction merges; compaction is normally the largest consumer, which is another reason to cap it.

Catch-up time decides how long a partition runs with fewer than three copies. Installing one 2.0 GB partition snapshot at an assumed throttled 50 MB/s takes 40 seconds plus the log tail. Replacing a failed node means re-creating its 32 partition replicas: 32 × 2.0 GB = 64 GB, which at an assumed 50 MB/s of aggregate throttled copy is about 21 minutes, and at 200 MB/s about 5 minutes. Raising the cap shortens the window of reduced redundancy at the price of foreground latency; the pace is a policy decision, and snapshot lag (how far a new replica's applied index trails the leader's) is the metric that shows it.

An SLO (service level objective) turns these dashboards into decisions. If Meridian promises 99.95% monthly availability for linearizable reads and writes, the error budget is 0.05% of a 30-day month, 43,200 minutes × 0.0005 = 21.6 minutes. A planned migration that pushes P99 past the objective, or an election that rejects writes for a few seconds, spends part of that budget, so risky background work is paced against the remaining budget rather than run whenever it is convenient.

What to watch

Compaction is background work that needs headroom. When pending bytes grow faster than compaction drains them, the number of files a read checks rises and tail latency rises with it; once free space falls under the target, compaction temp space is at risk. Throttle background tasks, alert on pending compaction bytes and free disk, and shed or reject writes by policy before the disk is full — an explicit failure, not a corrupted one.

• Track P50/P95/P99 per operation and commit latency per partition.
• Alert on under-replicated partitions, replication lag, compaction debt, free disk against the 60% target and repair age against tombstone retention.
• Cap background repair, compaction and rebalance bandwidth; show the foreground P99 impact.
• Throttle per tenant and expose stuck migrations.

Hot keys are not fixed by capacity. The hot key at 60,000 reads/s needs at least three serving replicas at an assumed 25,000 reads/s each (3 × 25,000 = 75,000, 80% utilised, little headroom). Options are explicitly stale follower reads, a TTL client cache that is optional and labeled, request coalescing, or splitting read-only data across keys. Writes to one key still serialize through one leader, and salting does not change that.

These sketches are conceptual: they omit batching, retries inside the replication layer, cancellation of in-flight requests and many membership details, and they are not an executable Raft implementation. Each is followed by a concrete timeline.

Put path

Timeline: the client sends evt-024; the router uses epoch 12 for partition 417; node-14 appends index 88213; node-52 flushes and replies; commit; apply produces v8; 200. If the leader crashes after commit but before replying, the retry reaches the new leader, finds the request id in the applied table and returns v8. If the leader crashes *before* a majority stored it, the new leader may or may not hold the entry: if it does, the entry is committed under the new term and the retry is deduplicated; if not, the retry applies it once. Either way the outcome is applied exactly once. Canceling the wait on a client timeout must not cancel the replication.

Read path

SSTable lookup

For an absent key and five SSTables the expected wasted reads are about 0.041 with the 10-bit filter described earlier.

Safe split or rebalance

Repair and replay

Conditional-update race

Timeline: key atlas/user42/settings is at v7. Writers X and Y both call Put(ifVersion=7). The leader logs X at index 88213 and Y at 88214. Applying 88213 yields v8 and 200 for X. Applying 88214 sees version 8, not 7, and returns 412 with currentVersion 8 to Y, who re-reads and retries with ifVersion=8. No client-side lock was needed and no replica disagrees about the winner. A client-side read-then-write would have let both overwrite v7 with their own v8.