Distributed Key-Value Store Interview Guide: Design & Follow-ups
Interviewers asking for a distributed key-value store are mostly checking whether you can say what an acknowledged write promises, what survives a failure, and what each stronger guarantee costs. This guide answers thirty-six questions in full, from write-ahead logs and LSM storage through partitioning, quorum replication, conditional writes, tombstones and hot keys to multi-region recovery, and adds a clarifying-question checklist, a 45-60 minute path, level-by-level expectations, a scoring rubric, common errors with corrections and a spoken whiteboard walkthrough. The fictional Meridian Profiles scenario and every number in it are teaching assumptions, not measurements.
What this interview is really probing
A distributed key-value store question is a test of whether you can say, for each promise the system makes, who enforces it, what survives a failure, and what you give up to keep it. Drawing a ring of nodes with arrows is the easy part; explaining what an acknowledgement means after a leader crash is the part that separates answers.
The guide below covers thirty-six questions, ordered the way a design conversation tends to unfold: scope and API, durability, storage engine, partitioning, replication and consistency, concurrency, data lifecycle, operations and finally geography. Every answer is written out in full, so each one can be read alone. The running example is Meridian Profiles, a fictional B2B platform that stores user preferences, account settings and tenant configuration; its tenants Atlas, Birch and Cedar, its keys such as atlas/user42/settings and its tracked write evt-024 exist only for teaching. All volumes, latencies and per-node budgets are labeled assumptions, not measurements and not data from any real company. Nothing here claims that any particular company asks this question.
The design that the answers defend is a reference design: keys are split into partitions, each partition is replicated three times and commits writes through a Raft-style replicated log with a single leader, and each replica stores data in a write-ahead log plus a log-structured merge tree. "Raft-style" is a conceptual model here, not a claim about any product. Where a leaderless, Dynamo-style design or multi-active regions would answer differently, the answer says so and keeps the two sets of guarantees apart.
The first five minutes and a 45-60 minute path
Before any box is drawn, spend a few minutes pinning the contract down. These are the clarifying questions that most change the architecture, with the assumption this guide makes if the interviewer says "you decide".
| Question to ask | Why it changes the design | Assumption used in this guide |
|---|---|---|
| Is this a cache, or the system of record? | A cache may drop or evict entries silently; a system of record must not lose an acknowledged write. | System of record. A write that has been acknowledged survives the loss of any single machine. |
| What operations are needed? | Point lookups hash well; range scans need ordered partitioning; transactions need coordination. | Put, Get, Delete, conditional write on a version, optional BatchGet, prefix scan only where data is range-partitioned. No joins, no cross-key transactions, no secondary indexes. |
| How big are keys and values? | Size drives memory, replication bandwidth and whether a value should live elsewhere. | Key up to 1 KiB, value up to 256 KiB, average record 1 KiB (teaching limits). |
| What read consistency is needed? | Linearizable reads cost a leader round trip; stale reads can be served by any replica. | Single-key linearizable by default; named weaker modes per request. |
| How many records and how much traffic? | Decides whether one machine could hold it. | 2 billion records; 200,000 reads/s average, 1,000,000/s peak; 40,000 writes/s average, 200,000/s peak (assumed). |
| How many regions, and are there residency rules? | Synchronous cross-region commits add intercontinental latency to every write. | One home region (EU) with an asynchronous US replica; EU-resident records stay in the EU. |
| What should happen when no safe answer exists? | This is the CAP choice for the affected operations during a partition. | Return an explicit unavailable error rather than guess; never a false success. |
A one-minute summary after the answers might sound like this: "We are building a durable key-value store, 2 billion records of about 1 KiB, mostly reads, with single-key linearizable writes in a home region. Records are hash-partitioned into virtual partitions; each partition is three replicas on separate failure domains with one leader and majority commit; each replica uses a write-ahead log and an LSM tree. Conditional writes are part of the replicated state machine. We prefer rejecting to guessing during partitions, replicate asynchronously to a second region and state the data-loss window honestly."
A realistic 45-60 minute path
| Minutes | Phase | What gets said and drawn | Board afterwards |
|---|---|---|---|
| 0-5 | Clarify | The questions above, plus non-goals. | Requirements list and a line of non-goals. |
| 5-10 | Capacity | 2.048 TB logical data, 40.96 MB/s average client ingest, per-node budgets, why this is throughput-bound rather than capacity-bound. | A corner of numbers, each marked as assumed. |
| 10-18 | Single node | In-memory map, why it loses data, write-ahead log with fsync, memtable, SSTables, Bloom filters, compaction. | One storage engine box with the write and read paths. |
| 18-27 | Partitioning | Why modulo hashing moves 75% of keys, virtual partitions, partition map with epochs, safe movement. | Router, partition map, 1,024 virtual partitions. |
| 27-38 | Replication | RF=3, leader, majority commit, term fencing, read modes, minority partition. | Replica group A/B/C with commit arrows. |
| 38-46 | Concurrency and lifecycle | CAS, idempotency keys and ambiguous timeouts, tombstones, TTL, repair, backups. | Version counter and a tombstone timeline. |
| 46-55 | Scale and regions | Hot keys, compaction pressure, tenants, EU home region and a lagging US replica, RPO and RTO. | Second region and an operations dashboard. |
| 55-60 | Wrap-up | Spoken summary, residual risks, what to measure first. | Nothing new. |
Adapting depth to the level
| Level | Expected shape | Depth signals |
|---|---|---|
| Mid-level | A working design: partitioning, replication factor, a write-ahead log, a clear read/write path. | Knows why a crash loses an in-memory map and why modulo hashing is a poor choice; names quorum and can compute a majority. |
| Senior | The above plus an explicit consistency contract and failure behavior. | Separates acknowledged from durable, explains fencing and ambiguous timeouts, sizes the cluster with assumptions, knows what compaction does to tail latency. |
| Staff | Trade-offs across alternatives and operations, with evidence. | Compares leader-based, leaderless and range-tablet designs, chooses between them for stated reasons, defines RPO/RTO honestly, plans migrations, testing and observability, and says what would change the decision. |
Scope, API and assumptions
These first three questions decide what the system is. Get them wrong and every later answer is built on the wrong promise.
Why build a durable key-value store instead of just using Redis (a popular in-memory data store) as a cache?
A cache and a system of record make opposite promises. A cache is allowed to forget: entries are evicted under memory pressure, expire on a timer and can be rebuilt from the real source. Meridian's settings records have no other source. If atlas/user42/settings disappears, the tenant's configuration is gone, so the store has to promise that an acknowledged write survives machine failure. That promise needs a write-ahead log flushed to stable storage before the acknowledgement, replication to other machines before the acknowledgement, and no silent eviction anywhere in the path. Redis is a fine fit for the cache role, and it can be configured with persistence (an append-only file with a chosen fsync policy) and replication, but its default posture is memory-first and its replication is asynchronous, so an acknowledged write can be lost on failover; options such as waiting for replicas to acknowledge narrow that window, yet they do not turn it into a consensus-committed log. If the question is purely latency for data that exists elsewhere, put a cache in front. If the data lives only here, pay for durability and say what it costs: higher write latency (a flush and a replica round trip), more hardware (three copies) and more operational surface.
- Could you front the durable store with a cache anyway? — Yes, and many systems do. The cache then holds copies with a TTL; the cost is staleness and invalidation, so reads that need the latest value must bypass it. (Further reading, not required: a cache stampede is what happens when many popular entries expire at once and every client rebuilds them simultaneously.)
- Is Redis never a database? — Not what is claimed. Redis can be used as a primary store when its persistence and replication settings meet the loss tolerance; the point is that the default settings and the asynchronous replication make that a deliberate configuration choice rather than a given.
What are the core APIs, and what does each return?
Four operations cover the product, plus two optional ones. Put(key, value, expectedVersion?, idempotencyKey) writes a value; with expectedVersion it becomes a conditional write that succeeds only if the stored version matches. It returns committedVersion and an ETag on success, 412 Precondition Failed on a version mismatch, 409 if the same idempotency key arrives with a different payload, 503 when no safe leader or quorum is available, and 429 when the tenant is throttled. Get(key, readMode) returns value, version and the read mode that actually served it; a key that never existed, was deleted or has expired returns 404 (a product may choose to distinguish deleted or expired from never-existed, but that is a decision, not a requirement). Delete(key, expectedVersion?) is a versioned write of a tombstone, not an erasure. BatchGet fetches many keys with no atomicity across them. Scan(prefix) is only offered where data is range-partitioned, because a hash-partitioned store scatters a prefix over every partition. A change stream (Watch) is advanced and optional. A request that times out has an unknown outcome: the client retries with the same idempotency key so a duplicate cannot apply twice. Deliberately absent are joins, cross-key transactions, secondary indexes and full-text search; adding any of them changes the architecture rather than extending it.
- Why return the version on every read? — It is what makes compare-and-set possible without a read-modify-write race: the client passes back the exact version it saw.
- Why does Delete take an expected version? — So that a delete issued against an old view cannot silently remove a newer value.
What assumptions do you make about key and value sizes and about consistency?
Sizes first, because they set cost. For the settings store the assumed limits are a key of at most 1 KiB and a value of at most 256 KiB, with an average record near 1 KiB. Those limits exist for a reason: very large values inflate replication traffic, memory for memtables and block caches, and compaction rewrites, and a 5 MB blob in the commit log delays everything behind it. Anything bigger belongs in an object store with only the pointer and hash in the key-value store. Keys are tenant-scoped composites like atlas/user42/settings, so the tenant prefix is part of the identity rather than a convention. On consistency the stated default is single-key linearizable reads and writes in the home region, which means every operation on one key appears to take effect at one instant between its request and response. Weaker modes are opt-in and named: follower-stale (labeled with a staleness bound or none), read-your-writes (using a session token carrying a version floor) and eventual (remote regions). Nothing spans keys: two Puts to different keys have no ordering guarantee relative to each other beyond what the client imposes. Those are limits the interviewer should hear out loud, because each one is an explicit non-goal that keeps the design tractable.
- What if a customer wants a multi-key update to be atomic? — That is a different product, with transactions and cross-partition coordination such as two-phase commit, and it needs its own design; the reference store would push the problem to the application or to a transactional store.
Capacity and what an acknowledgement promises
Numbers come before architecture here, but they are inputs to be challenged, not facts. Each calculation below states what it assumes.
Do the capacity math for 2 billion keys.
Start with data. Two billion records at an assumed average of 1,024 bytes each is 2,048,000,000,000 bytes: 2.048 TB decimal, about 1.86 TiB, for a single logical copy. With a replication factor of 3 the raw record copies come to 6.144 TB decimal (about 5.59 TiB). That is a count of copies, not a disk purchase. Provisioned disk is a separate worksheet: add an assumed 5% for indexes, Bloom filters and metadata (6.4512 TB), then keep volumes under an assumed 60% full so compaction, snapshots and repair have room, giving 6.4512 ÷ 0.6 = 10.752 TB, roughly 112 GB per node over 96 nodes, before write-ahead logs and backups. That is small. Traffic is what sizes the cluster. Assume 40,000 writes/s on average and 200,000/s at a short peak, 200,000 reads/s average and 1,000,000/s at peak. Each write is applied on 3 replicas, so peak write applies are 600,000/s. If one node sustains an assumed 20,000 write applies or 25,000 point reads per second, the peak needs 600,000 ÷ 20,000 = 30 plus 1,000,000 ÷ 25,000 = 40, so 70 node-equivalents; divide by a 0.7 utilization target and you get 100, against 96 planned, which is acceptable for a short burst. The conclusion to say aloud is that this workload is throughput-bound, not capacity-bound, so adding disk would not help but adding nodes would.
- How many partitions, and how big is each? — Assume 1,024 logical partitions: 2.048 TB ÷ 1,024 = 2.0 GB each, about 1.95 million records. With RF=3 that is 3,072 replicas, or 32 per node on 96 nodes, 64 GB of replicated data per node.
- Is the peak figure realistic for sizing? — It is a planning illustration. Sizing for a burst that lasts seconds costs money all year, so real teams combine headroom with load shedding and per-tenant throttles.
What exactly does an acknowledged write promise?
It promises that the write is committed, and committed has a precise meaning: the write is stored in the write-ahead log, flushed with fsync on a majority of the three replicas, and the leader has advanced its commit index past it. From then on, any replica that can become leader holds it, so losing one machine, or the leader itself, cannot lose it. It does not promise that every replica has applied it to its tables yet, that follower reads will show it immediately, or that a remote region has it. It also does not hold if the storage stack lies about flushes, since a disk or controller with a volatile write cache that ignores flush requests can lose data the software believed was on stable media. And an acknowledgement can be lost in transit: if the leader commits and the response never arrives, the client sees a timeout, which means unknown outcome rather than failure. Putting this together, the sentence to say is: "Acknowledged means durable on a majority of replicas within the home region and visible to linearizable reads; replicas elsewhere are asynchronous." Anything weaker should be a named mode, not a silent default.
- What if only two of three replicas are up? — They are a majority, so commits continue; the third catches up when it returns. If only one is up, writes are rejected with 503.
- Does 'durable' mean the data survives losing a data center? — Only if the three replicas sit in separate failure domains such as availability zones; a region-wide loss is a separate question about the asynchronous remote replica.
Why does the write-ahead log come before the response?
Because memory forgets. Imagine the store applies a Put to an in-memory map and replies OK. Two milliseconds later the process is killed. The client was told the write succeeded, the data was only in RAM, and nothing remains: that is a lost acknowledged write, which breaks the system-of-record contract. The log fixes it by changing the order of operations. Append a record {seq, op, key, value, checksum} to a sequential file, make it durable, then apply it to memory, then reply. If the process dies afterwards, restart reads the log and rebuilds the state. Two details matter and are often blurred. Appending is not enough: a write() call usually only copies bytes into the operating system's page cache, and the data is durable only after fsync (or a file opened with synchronous semantics) returns. And fsync is slow, which is why real engines batch: with a 1 ms commit window and an assumed 6,250 writes/s on a node, about 6 records share each fsync, trading a millisecond of latency for far fewer flushes. The log is sequential appends, which disks do far better than random writes, so the cost is a predictable one rather than a disastrous one.
- Could we skip fsync and replicate instead? — If three machines in separate failure domains hold the write in memory, a single crash cannot lose it, but a correlated failure such as a power event can, so this trades durability for latency; it is a legitimate choice only if stated, and it is not the reference behavior.
- What if the machine crashes mid-append? — The tail record may be torn. Each record carries a checksum and length, so recovery stops at the first record that fails verification; since that write was never acknowledged, discarding it is correct.
Walk through replay after a crash.
Take a node whose process dies after acknowledging evt-024. On restart it knows only its disk. It first loads the newest durable snapshot or the set of SSTables, each of which records the highest log sequence number it already contains, say 9,800. Then it opens the write-ahead log and reads records from sequence 9,801 onward, verifying each checksum, and re-applies each operation to a fresh memtable in order. When it meets a record whose checksum fails or whose length runs past the end of the file, it treats everything from that point as a torn tail and truncates it: those bytes were never fsynced as a complete record, so no client was told they were durable. Replay must be idempotent, since a crash during recovery means doing it again; applying the same sequence number twice has to produce the same state, which is why records carry sequence numbers and versions. Once replay completes the node holds every complete record it had durably logged, which includes everything it acknowledged as committed but may also include a few entries the group never committed. If it is a replica, it then asks the leader for the entries committed after its last index, and any uncommitted tail that conflicts with the leader's log is overwritten. The surprising piece is the order: recover from local disk, then catch up from peers, never the reverse, because the local log may hold entries that the group has since decided differently.
- How long can replay take? — Proportional to the log since the last flush; that is why memtables flush to SSTables and old log segments are discarded once the data they cover is durable elsewhere.
- What if the node's log has entries the new leader does not? — Uncommitted entries from an old term are overwritten by the leader's log; they were never acknowledged as committed.
The storage engine on each replica
Each replica is a single-machine database in its own right. These four questions are about what happens between the write-ahead log and the disk, and why tail latency often originates there.
LSM tree or B-tree for the storage engine?
For this workload, an LSM tree, with the reasons attached and the exceptions named. A B-tree updates pages in place: a write finds the right leaf, modifies it, and relies on a log for crash safety. Reads are predictable (a few page fetches), and a range scan walks adjacent leaves. The price is random writes and page splits. An LSM tree turns every write into sequential work: append to the log, insert into an in-memory sorted memtable, and when it fills, flush it as an immutable sorted file (an SSTable). Reads may need to look in several files, so Bloom filters skip files that cannot hold the key, and background compaction merges files to bound how many exist. Meridian reads about five times more than it writes (200,000 against 40,000 per second on average), but 40,000 writes/s of 1 KiB records is a sustained flow into a 2 TB dataset, so cheap sequential writes still matter and Bloom filters plus a block cache keep the read side acceptable. Versioned records with tombstones fit the append-only model naturally. The costs are real and should be said: reads can touch several levels, compaction rewrites data repeatedly (write amplification) and competes with foreground I/O, and space amplification during merges needs headroom. A B-tree is the better choice when reads dominate, when latency must be tightly predictable, or when the working set is small and in-place updates are cheap. Neither choice is wrong in general; the stated workload decides, and the choice should be revisited by measurement, not by taste.
- Where does an LSM tree's write amplification come from? — Each record is written once to the log, once at flush, and again each time compaction rewrites it into a lower level. As an illustrative figure only, assume a combined factor of 10; with 204.8 MB/s peak client ingest, three replicas and that factor, the cluster writes about 6.1 GB/s to disks, roughly 64 MB/s per node over 96 nodes. Measure the real factor.
- Which has better space usage? — It depends on strategy; leveled compaction keeps space amplification low at the cost of more rewrites, while size-tiered compaction does the opposite.
Why can compaction hurt P99 latency, and what do you do about it?
Compaction competes with foreground requests for the same disk bandwidth and CPU. Picture a node serving reads with a P99 of 5 ms. Several level-0 files accumulate because flushes outpace merges, and now each read must check more files; at the same moment a large merge is saturating the disk with sequential reads and writes, so the reads that do miss the cache queue behind it. Median latency barely moves, but the slowest one percent doubles or worse. If compaction falls too far behind (compaction debt), the engine eventually has to slow or stall writes to let it catch up, and that appears to users as a sudden write latency cliff. Mitigations are mostly about bounding and observing the background work: rate-limit compaction I/O so foreground traffic keeps a reserved share, schedule heavier merges off-peak where the strategy permits, keep the number of files per level in check, give compaction its own I/O priority, and alert on pending compaction bytes and level-0 file counts before reads degrade. A larger design lever is spreading replicas so that not every replica of a partition compacts at once, letting a read-quorum or leader choose a healthy node. The thing to avoid is treating compaction as free background noise: it is a second workload with its own capacity budget.
- Should you just throttle compaction hard? — Throttling protects reads but lets debt grow; if debt grows long enough you hit a write stall anyway, so the throttle has to be paired with enough disk throughput and an alert on debt.
- Can we move compaction elsewhere? — Some designs separate storage from compute so compaction runs on different hardware; that adds network cost and a different failure profile, so it is a change of architecture rather than a tuning.
What is read amplification?
It is the ratio of work done to find data to the work that would be needed in an ideal layout, usually counted as the number of files or blocks touched per lookup. Imagine a key that is not present and an LSM tree with 5 SSTables that could cover its range. Without help the engine would read an index block and maybe a data block from each, which is 5 or 10 disk reads for a lookup that returns nothing. A Bloom filter per SSTable fixes most of this. With 10 bits per key and 7 hash functions the false-positive rate is about 0.82%, so 5 files cost an expected 5 × 0.0082 = 0.041 needless reads, and a correctly built filter has no false negatives, so a present key is never skipped. The memory is modest: 1,953,125 keys × 10 bits is about 2.44 MB per partition replica, or about 78 MB per node at 32 replicas. A key that is present still has to be found in the newest file that contains it, so reads check the memtable first, then SSTables newest to oldest; more levels or more overlapping files means more amplification. Compaction reduces overlap, which is why it is both a cost and a read optimization. A B-tree has its own amplification, typically one page fetch per level, which is why the comparison has to be made per workload.
- Do Bloom filters help range scans? — Not for ranges. A filter answers membership for a specific key; prefix filters exist for some engines, but a general range scan still merges iterators over all relevant files.
- How do you cut read amplification further? — Fewer overlapping files via leveled compaction, a larger block cache, and keeping hot keys in memory; each costs write amplification, memory or both.
How do you recover from disk corruption?
First decide what kind of corruption, because the responses differ. A torn write at the tail of the log after a crash is normal and handled by checksums during replay. Silent corruption of an SSTable block is different: a block checksum fails on read, and the engine must not return the bytes. What happens next depends on replication. On one replica, mark the file or the node as suspect and serve the request from another replica, which is possible because a committed record is held by a majority and so is normally on both other replicas and certainly on at least one of them; the corrupt replica is not trusted to repair itself in place. The standard recovery is to discard the damaged replica's data and rebuild it from the leader or another healthy replica, by installing a snapshot of the partition and replaying the log from the snapshot's index. That takes time proportional to the 2 GB partition (in the assumed sizing) and consumes network and disk on the source, so it is rate-limited and throttled like any repair. During the rebuild the partition runs with two replicas, which still commit, but a second failure would block writes, so the alert for under-replicated partitions matters. If every replica of a key is corrupt (a software bug writing bad bytes through the whole group), replication does not help, and recovery comes from backups and point-in-time restore. That is the reason checksums on read, periodic scrubbing and tested backups all exist as separate layers.
- Does replication protect against corruption? — Against hardware corruption on one replica, yes, because the others hold good copies. It propagates logical corruption: a bad application write is faithfully replicated to all three.
- How do you detect corruption before a user reads the data? — Background scrubbing reads blocks and verifies checksums on a schedule, and repair compares replicas by hash trees or digests.
Partitioning and moving data
Once one machine cannot hold or serve the workload, keys must be divided among machines, and the division has to survive membership changes.
How is a key routed to the node that owns it?
Through a layered lookup with a stale-tolerant fast path. The client library or router hashes the key (the tenant prefix is part of the hashed key) into a position in a fixed space, say 0 to 2^64, and maps that position onto one of 1,024 logical partitions. A partition map then says which replica group currently owns each partition: for atlas/user42/settings it might be partition 417, owned by replicas on nodes 12, 40 and 77, with node 40 as leader in term 9. Each entry carries an epoch that increments whenever ownership changes. The router caches the map and sends the request to partition 417's leader with the epoch it believes is current. If ownership has since moved, the receiving node compares epochs, sees the router is stale, and answers with a retryable WRONG_EPOCH or a redirect pointing to the newer map; it must not serve the request as though it were still the owner, or two nodes would both act as authoritative for one key. The map itself is small (1,024 entries, a few hundred KB) and is kept in a strongly consistent metadata store, cached aggressively by routers. The important property is that routing may be stale but ownership is checked at the node, so staleness costs a retry rather than correctness.
- Where is the partition map stored? — In a small strongly consistent metadata service, for example a Raft-backed store like etcd in some deployments; it is read by many but written rarely.
- What if the router cannot reach the metadata service? — It keeps routing with its cached map, and nodes keep rejecting wrong-epoch requests; only changes to ownership stall.
What breaks with modulo sharding?
Adding or removing a node changes the divisor, and with it nearly every key's destination. Suppose a hash h places a key on node h mod 3 with 3 nodes and you add a fourth, now h mod 4. A key stays put only if the two remainders agree. The remainders agree only when h mod 12 is 0, 1 or 2, so 3 of every 12 hash values stay: 25% of keys stay and 75% move. With 2.048 TB of logical data, that means 1.536 TB crossing the network, or 4.608 TB counting all three replica copies, while the cluster is also serving traffic, which is exactly when you added the node because it was running hot. During the migration, routers and nodes disagree on where each key is, so you need either downtime or a very careful dual-read scheme. Consistent hashing, or fixed virtual partitions, changes the picture: when a fourth node joins three, only the share it should own moves, about 25% or 0.512 TB logical, and everything moves to the new node, not between existing ones. That is still real data to copy, so the claim is not zero movement, but it is the minimum movement. Modulo is fine for a fixed set of shards that never changes; the moment the node count is expected to change, a fixed-partition layer between keys and nodes is the fix.
- Is consistent hashing alone enough? — It reduces movement but can produce uneven load unless each node holds many virtual positions; fixed partitions with a map give the same effect with explicit control.
- Why not just rehash everything offline? — That needs downtime or a write freeze proportional to the data size.
When would you choose range partitioning?
When the product needs ordered access. Hash partitioning spreads keys evenly but destroys adjacency, so a request for all keys with prefix atlas/user42/ must ask every partition. Range partitioning keeps keys sorted and assigns contiguous ranges (tablets) to servers, so a prefix scan touches one or a few ranges in order. That is the model used by Bigtable-style systems, which describe tables as sorted key-value maps; the same property supports efficient range deletes, time-ordered keys and locality when related rows share a prefix. The costs are skew and hotspots. If keys are sequential timestamps or auto-incremented ids, every new write lands in the last range, so one server takes all the write traffic; the usual answers are to prefix keys with something that spreads them (a hash bucket or tenant) at the price of losing global ordering, or to split ranges automatically when they grow or heat up. Range maps also need a split-and-merge mechanism and a metadata layer that records boundaries. In the Meridian design lookups are mostly point reads by tenant/user key, so hash partitioning is the default, and the system offers prefix scan only for a range-partitioned keyspace where a feature really needs it. If the interviewer says analytics-style scans matter, switch the answer and defend range tablets with automatic splits.
- Can you get both even load and range scans? — Partially: a composite key like
tenantBucket/userId/…keeps a user's data contiguous while spreading users, so scans are efficient within a user but not across users. - What do hot ranges look like? — One tablet with a disproportionate share of requests; splitting helps unless the heat is one key.
What is a virtual node, or a tablet?
A virtual node is a small unit of ownership: instead of giving each physical machine one big slice of the keyspace, the keyspace is cut into many pieces and each machine owns several. In the teaching sketch 12 virtual partitions are shared by 3 machines, 4 each. Add a fourth machine and the system hands over 3 partitions in total (one from each of the original three, or any arrangement that ends 3-3-3-3), so exactly 3 of 12, or 25%, move and nothing else is touched. Without virtual nodes, a node would own one contiguous range of hash space and adding a node would split one neighbor's range, giving it load while the others stay unchanged. Many small pieces also make rebalancing granular (move one 2 GB partition rather than a quarter of a machine), spread a failed node's load across many survivors instead of dumping all of it on one neighbor, and let heterogeneous machines own different numbers of pieces. A tablet in range-partitioned systems is the same idea with a contiguous key range as the unit, split when it grows. In the sizing here, 1,024 logical partitions of 2.0 GB give 32 replicas per node on 96 nodes. Too many pieces cost metadata and per-partition overhead such as separate logs; too few make moves coarse, so the number is a design choice to revisit.
- Why fix the partition count in advance? — It keeps the key-to-partition mapping stable; only the partition-to-node assignment changes. Splitting a partition is a heavier operation and is done deliberately.
- What limits the partition count? — Per-partition overhead (logs, memtables, election timers) and the size of the map.
How are partitions moved safely?
By treating a move as a change in membership that the data has to catch up with, in steps that keep exactly one authority at each moment. Suppose partition 417 moves from node 12 to a new node 90. First the control plane records the intent and adds node 90 as a non-voting learner of the partition's replica group. The learner receives a snapshot of the data and then the log entries after the snapshot until it has caught up, which is rate-limited so it does not starve foreground traffic. When the lag is small, the group is reconfigured through the replicated log so that node 90 becomes a voting member, and node 12 is later removed; the partition map gets a new entry with a higher epoch. Throughout, writes keep going to the leader and are replicated to the learner as well, so there is no write freeze. If the old leader is being replaced, leadership transfers first to a node that stays. Any request that arrives at a node with an old epoch gets WRONG_EPOCH and retries after fetching the new map, so no key has two active owners. Danger points are a learner that never catches up because writes outpace the copy, a move canceled halfway (the old replica must stay valid until the new one is confirmed), and moving too many partitions at once; the operation caps concurrent moves and rate-limits copying. The failure mode to design against is the stuck migration, which is visible on a dashboard and abortable without data loss.
- Why not copy the data and flip a switch? — Writes that arrive during the copy would be lost or require a freeze; catching up through the log removes that gap.
- What if the destination dies mid-move? — The source remains the authority; the move is retried or abandoned, and the map epoch does not change until the group has been reconfigured.
Replication, quorum and consistency
This is the center of the interview. The questions are ordered so that each one fixes a weakness left by the one before.
Why replication factor 3?
One copy means a single disk or machine failure loses data and makes the partition unavailable until it returns. Two copies survive one failure but cannot form a safe majority: with two replicas, a majority is two, so losing either one blocks commits, and if the group tries to continue with one it risks divergence in a partition. Three is the smallest group where a majority (two of three) survives one failure, so writes keep committing and a replacement can be built from the survivors. Larger groups buy more tolerance at more cost: five replicas have a majority of three and survive two failures, but every write now needs three durable copies and storage and bandwidth grow by two-thirds over RF=3. The arithmetic for the reference: 6.144 TB of record copies against 2.048 TB logical. The common misreading is that three replicas means any two machines can be lost and writes still succeed. They cannot: with two of three gone, no majority exists, so commits stop and the strongly consistent read modes become unavailable; the data is probably still on the surviving machine, but the system cannot prove it is the latest. Placement matters as much as the count. Three replicas on one rack or one power feed share a failure, so replicas are spread across separate failure domains such as availability zones.
- When would you choose RF=5? — For data where tolerating two simultaneous failures justifies the cost, or for metadata like the partition map; most value data stays at 3.
- Does RF=3 also give three times the read capacity? — Only if reads may be served by followers, which brings staleness questions; linearizable reads go through the leader or a confirmed read index.
How does consensus commit differ from replication factor alone?
Replication factor says how many copies exist; it says nothing about when a write counts as done, who may accept it, or what happens when copies disagree. Imagine three replicas and a rule 'write to all, acknowledge whenever the leader has it'. The leader crashes after acknowledging but before copying; a follower with no knowledge of the write becomes leader, and the acknowledged write is gone. Make it 'acknowledge after any one follower has it' and a partitioned group can elect a leader without that follower. A consensus protocol such as Raft closes these holes with two rules. Every write is an entry in a replicated log with a term and an index, and the leader counts it committed only when a majority has stored it. A new leader can be elected only by a majority, and a voter refuses a candidate whose log is less up to date than its own, so any majority that elects a leader overlaps the majority that held a committed entry, and the new leader must already have it. Terms fence stale leaders: an old leader that comes back sees a higher term and steps down. In evt-024 for atlas/user42/settings: leader A appends it at index 5801 in term 9, ships it to B and C, B acknowledges, so A has 2 of 3 and commits, then replies. C is lagging and gets it later without affecting the outcome. The model here is Raft-like and conceptual; RF=3 on its own is not consensus.
- Does the leader wait for all three acks? — No, for a majority; that is why one slow replica does not slow every write.
- What does a follower do with an entry it holds that was never committed? — If a new leader has a different entry at that index, the follower's version is overwritten; since it was never committed it was never acknowledged.
What is linearizability, and what does it buy you?
Linearizability is a recency and ordering guarantee for operations on a single object: each operation appears to happen atomically at some instant between when the client sent it and when the response came back, and the order of those instants is consistent with real time. Concretely, if Put(settings, v8) returns to one client, then any Get started after that return, by any client, must see v8 or something later; it can never see v7. This is what lets people reason about a key like a variable in a single-threaded program. It is not serializability, which concerns multi-object transactions; a linearizable store can still offer no ordering between two different keys. And it is not the same as durability or strong-sounding marketing: it is a precise statement about visible order. The reference store provides it per key in the home region through the leader. Reads are only linearizable if the leader proves it is still the leader before answering, for example by confirming with a majority (the read-index technique), or by holding a lease that depends on bounded clock drift, because a deposed leader answering from local state would return stale data. The price is a round trip to the leader, and unavailability if no safe leader exists.
- Is a read from the leader automatically linearizable? — No; a leader that has been partitioned away and does not know it yet could return stale data, which is why the read confirms leadership or relies on a lease.
- Why would anyone want less? — Latency, locality and availability: a nearby follower answering in 1 ms beats a leader round trip when slightly old data is fine.
How can a follower be read safely?
Safely means the client knows what it is getting. A follower that has applied entries up to index 5799 while the leader committed 5801 may return the old value of a key, so the response must say it is a follower-stale read, and ideally by how much. There are three ways to make this useful. First, bounded staleness: the follower knows its apply lag against the leader's commit index or a heartbeat timestamp, and refuses to answer if it is more than, say, 2 seconds behind (an assumed bound). Second, read-your-writes through a session token: after a Put returns committedVersion 8, the client sends that version as a floor, and a follower answers only if it has applied at least that point, otherwise it waits briefly or forwards to the leader. Third, a follower read that first asks the leader for the current commit index and waits until it has applied that far, which restores linearizability at the cost of a round trip and is often cheaper than shipping the whole read to the leader. Pure unlabelled follower reads under write load give you old values at times, such as showing a tenant's previous settings right after saving; that is acceptable for a dashboard and wrong for an authorization check. Tag the mode on every response so callers cannot mistake it for the strong one.
- What does the user see with plain follower reads? — They may save a setting and then see the old one on the next page load; read-your-writes removes that specific surprise without full linearizability.
- How does a follower know its lag? — From the commit index and timestamps carried in the leader's heartbeats.
What happens during a minority partition?
Consider replicas A (leader), B and C, and a network split that isolates A from B and C. Each side reacts by what it can prove. B and C cannot hear A's heartbeats; after an election timeout (the Raft paper suggests randomized values such as 150-300 ms, implementations differ) one of them becomes a candidate, gets the vote of the other, and is elected leader in term 10. They now hold a majority of two and continue committing writes. Isolated A is in the minority. It still believes it leads term 9, but it cannot reach a majority, so any write it receives stays uncommitted and is never acknowledged; after a leader-lease or check-quorum timeout A steps down and answers requests with an unavailable error, 503. Clients that can reach only A therefore see failures, while clients that can reach B and C see a working store: availability for the majority side and consistency for everyone, which is the CAP trade-off made explicit for the minority. When the network heals, A sees term 10, steps down and discards its uncommitted entries, and catches up from the new leader. If the split is three-way (one replica each) nobody has a majority and writes stop everywhere until it heals. A linearizable read from A in the minority must also be refused, since A cannot confirm it is still leader. No false success is ever returned, which is the property to verify in a test.
- What if a client talks to the old leader and gets a timeout? — It treats the outcome as unknown and retries with the same idempotency key against the current leader.
- Why not let the minority keep accepting writes? — Because it would produce a second history for the same key, which would have to be merged or discarded after healing; that is the leaderless design's choice, not this one's.
Can R+W>N guarantee linearizability?
No. It is an overlap guarantee, and overlap is necessary but nowhere near sufficient. In a Dynamo-style leaderless design with N=3 replicas, a write acknowledged by W=2 and a read that consults R=2 must share at least one replica because 2 + 2 > 3. That means the read will contact a replica that saw the latest acknowledged write, which is useful, and the read can compare versions and return the newest. But several things break the step to linearizability. Sloppy quorums and hinted handoff let writes land on stand-in nodes outside the usual replica set, so the overlap argument no longer refers to the same nodes. A write that failed partway may have reached one replica and not be acknowledged, so a later read may or may not return it, and two successive reads can disagree. Concurrent writes to the same key need a conflict rule (last-write-wins with clocks that can skew, or version vectors that return siblings), so there may be no single latest value. Membership changes alter the replica set mid-flight. Without a single ordering mechanism such as a leader and a replicated log, there is no point in time at which a write 'happened' for everyone. So say it as: R+W>N is quorum-overlap intuition for finding recent data; linearizability requires additional machinery such as consensus per key, and the leaderless design should be described as eventually consistent with tunable recency. Both designs are legitimate if their guarantees are not blended.
- Can you make a leaderless store linearizable for some operations? — Some systems add lightweight consensus or compare-and-set on top for specific operations; the guarantee then belongs to that operation, not to the quorum arithmetic.
- Which workloads suit the leaderless model? — Those that value write availability under partitions and can merge or tolerate conflicts, such as carts or counters with a merge rule.
Concurrent updates and ambiguous outcomes
Two clients, one key, and a network that can lose the answer: these two questions are about not losing or doubling updates.
How does compare-and-set avoid lost updates?
Start with the bug. Two admins open Atlas's settings at version 7. Admin one changes the timeout and saves; admin two changes the logo and saves. With a plain Put, whichever arrives second overwrites the whole record, and the first change disappears with no error: a lost update. Reading the current value, comparing in the application and then writing is no better, because the check and the write are separate steps and another write can slip between them. Compare-and-set puts the check inside the write. Each admin sends Put(atlas/user42/settings, newValue, expectedVersion=7). Both requests reach the partition leader, which appends them to the replicated log in some order, and the state machine applies them in that order. The first applied sees version 7, matches, writes version 8 and succeeds. The second is applied against a record at version 8, sees a mismatch and returns 412 Precondition Failed with the current version, without changing anything. That admin re-reads, merges the logo change into the version-8 value and retries with expectedVersion=8, giving v9. Because the comparison is a command in the replicated log and every replica applies it in the same order, exactly one writer wins and all replicas agree which. This only works because a key has a single serialization point in the leader's log; a leaderless design needs either a consensus-backed conditional write or version vectors with sibling resolution. The cost is retries under contention, and a hot key with many conditional writers can spend much of its throughput on 412s.
- Why is the version on the record, not a timestamp? — A monotonically increasing version assigned in the log order is exact; wall-clock timestamps can tie or reorder under clock skew.
- What does a client do on 412? — Re-read, merge or ask the user, and retry; a blind retry of the same payload would reintroduce the lost update.
How do idempotency keys handle an ambiguous timeout?
A timeout tells the client nothing about whether the server acted. Say the client sends Put with idempotency key k-7c1 and the leader commits it and replies, but the reply is lost; or the request never arrived. From the client's side the two are identical. Retrying a plain Put is harmless only if the operation is naturally idempotent, which a conditional write is not always (the first attempt may have moved the version to 8, so the retry with expectedVersion=7 now fails with 412 even though the client's change did apply), and an operation like 'append' or 'increment' would be applied twice. The fix is that the store remembers the outcome. The idempotency key and a hash of the request are recorded as part of the replicated state machine when the write commits, along with the result. A retry carrying the same key and the same payload returns the stored result instead of applying again. A request with the same key but a different payload is an error, 409, because the client has reused a key. The record cannot be kept forever, so retention is a documented policy, for example 24 hours (an assumed value), and the client contract says retries must happen within that window. The dedup record must live with the data it protects: if it were held only on one node and the leader changed, a retry could be applied twice. A client that gets 503 or a timeout therefore retries with the same key, with backoff and jitter so retries do not become a storm.
- Do reads need idempotency keys? — No; a Get has no effect, so retrying it is safe.
- What if retention expires before the retry? — The write would be treated as new; the contract limits retries to the window, and for conditional writes the version check provides a second line of defense.
Deletes, expiry and backups
Data that goes away is harder than data that arrives. These three questions are about removing records correctly while some replicas are absent, and about restoring to a known good point.
What are tombstones, and how does a deleted value come back?
A delete cannot simply erase the record, because the store is replicated and LSM-structured: older versions of the key sit in other SSTables and on other replicas, and a read that sees no entry would fall through to an older file and find the value again. So a delete writes a new, versioned marker: a tombstone saying that this key is deleted as of version 12. Reads that find the tombstone first return 404 and stop looking. Compaction can drop the shadowed older versions and, only much later, the tombstone itself. Resurrection is what happens if the tombstone is dropped too soon. Timeline: replicas A, B and C hold atlas/user42/settings at v11. C goes offline. The delete is committed on A and B as a tombstone at v12. Compaction on A and B then purges the tombstone after the retention period has been passed, and C returns after that, still holding v11. In a design that repairs by comparing replicas, C's v11 now looks like the only version of the key, and repair copies it back to A and B: the deleted settings are alive again. The rule is that tombstone retention must exceed the longest time any replica can be offline or unrepaired before it is brought back or rebuilt. As a teaching value, 10 days is used here (Cassandra's documented default for gc_grace_seconds is 864,000 seconds, 10 days). A replica that has been away longer than the retention must be wiped and rebuilt instead of rejoining with old data. In the leader-based reference design, a replica catching up through the log or a snapshot receives the tombstone and cannot resurrect anything, so the failure mostly lives in repair paths, restores and leaderless designs.
- Can the tombstone be removed immediately if all replicas are healthy? — Not safely; you cannot easily prove no replica, backup or in-flight repair still holds an older version, so retention is a time bound.
- What does a delete mean for storage? — Space usually increases at first, since a tombstone is added; it falls after compaction.
How does TTL interact with clock skew?
A TTL is a promise about time, and machines disagree about time. Say a settings override is written with expiresAt = 12:00:00. If expiry is evaluated independently by each replica using its local clock, a replica whose clock runs 2 seconds ahead stops returning the value at 11:59:58, and a replica that is 2 seconds behind returns it until 12:00:02. During that gap two reads of the same key can disagree, and a linearizable read can appear to go backwards. The way to reduce this is to make the expiry decision part of the replicated, ordered state. The leader assigns the write's timestamp (or the leader's lease/time source is used for expiry), the expiry deadline is stored with the record, and reads on the leader compare against one clock. Better still for strict cases, expiry becomes an explicit replicated operation, a tombstone written by the leader at or after the deadline, so all replicas converge on the same visible state at the same log position. Followers are allowed to hide a record past its deadline, but the guarantee is documented as bounded by the skew. Physical removal then works exactly like a delete: the tombstone persists until compaction past the retention period. Operational controls: monitor clock offset (NTP or PTP health), alert when drift exceeds the tolerance (an assumed 100 ms, say), and refuse to run a leader whose clock is outside it. Say plainly that TTL granularity is bounded by skew; a TTL of 200 ms is not honest in a system with a 100 ms drift budget plus a replicated commit, because the uncertainty is as large as the lifetime; a TTL of an hour is unaffected.
- Does expiry free disk immediately? — No. Expired data is logically hidden at once, and physically removed only when compaction reaches it, so space is reclaimed later.
- What about legal erasure timelines? — The physical removal bound includes compaction and backup retention; with an assumed 35-day backup retention a deleted record can remain in backups for up to 35 days, so the policy has to state this rather than promise instant erasure.
How do you back up and restore safely?
Think about what a restore is for: a bad deploy that corrupted values, an operator who deleted a tenant, a bug that wrote garbage through all three replicas. Replication does not help with those, since it copies the damage. The backup has two parts. Periodic consistent snapshots of each partition (SSTable files are immutable, so a snapshot is cheap to take with hard links or a manifest of file names, and can be copied out in the background), and the replicated log shipped continuously to object storage so that you can replay forward from a snapshot to a chosen point. Restore picks a target time, say 14:03:20, selects each partition's latest snapshot before it, and replays logged entries up to the target. Because partitions snapshot at different moments, the result is consistent per key and per partition but not a cross-key transaction; if the product needs cross-partition consistency, the restore tooling needs a common cut, and the product's non-goal of multi-key transactions should be remembered. Restore into a separate cluster first, verify a sample (checksums, counts, a handful of known tenants) and then either switch traffic or copy back the affected keys. Two cautions: a restore can resurrect deleted data, so any tenant deletion or erasure requests since the snapshot must be replayed afterward, and a backup you have never restored is a hypothesis. Schedule restore drills and record how long they take, because that duration is your real RTO for logical corruption.
- Why not rely on cross-region replication as the backup? — It replicates deletes and corruption as faithfully as good writes, so it covers a site loss but not a logical error.
- How long do you keep backups? — A policy decision balancing recovery needs against cost and erasure obligations; 35 days is used as an example only.
Scale, tenancy and operations
Production questions: what happens when load is lumpy, when the fleet changes, when customers share hardware, and how anyone knows it works.
How do hot keys affect capacity?
Averages hide them. In a simplified model with 100 partitions and 200,000 reads/s, an average partition serves 2,000 reads/s. Now atlas/feature=checkout_policy receives 60,000 reads/s. That is 30 times the average partition, 30% of average fleet reads from one key, and 6% of the 1,000,000/s peak. The partition that holds it, and in the linearizable path its leader, is far beyond its neighbors even though the cluster as a whole is idle. With an assumed 25,000 point reads/s per replica, 60,000 reads/s need at least 3 serving replicas (2.4 rounded up), and then those three run at 80% with almost no headroom. Adding nodes does not help, since the key still lives in one partition. The tools are specific. Serve labeled stale reads from followers so all three replicas share the load. Put a short-TTL cache in the client or a near cache, accepting staleness. Coalesce identical in-flight requests so a thousand concurrent reads cause one backend read. For read-only data, make the value available under several derived keys and have clients pick one at random. Writes are different: they must be serialized through one leader, so salting or fan-out does not speed up writes to a single hot key; the fix is to reduce write frequency, batch updates, or change the data model so unrelated updates are not stored under one key. Detect it by per-key and per-partition request metrics with top-k reporting, and apply per-tenant limits so one tenant's hot key cannot starve others.
- Is splitting the partition a fix? — Not for one key; a split divides the keyspace, but the hot key still lands in exactly one half.
- What does a cache add to the risk? — Staleness and a stampede when the entry expires; coalescing and jittered TTLs reduce it.
Why is adding nodes not free?
Because the cluster does more work in order to get less loaded. Adding a node does nothing until partition replicas are moved onto it, and moving them means copying data: for the fourth node joining three, about 25% of the logical data, 0.512 TB, has to be read from existing nodes, sent over the network and written, plus the log catch-up, all while those same nodes serve production reads and writes. That extra read, network and write load is the reason a cluster that is already at its limit can get worse for a while after a scale-out, which is the wrong time to find out. The copy has to be rate-limited, with a cap on concurrent moves, and finishing it can take hours at rates that do not hurt foreground latency. The move also touches the metadata path: new map epochs, router cache refreshes, learners, leadership transfer and brief failovers of leadership, each a chance for a retry blip. After the move, the new node starts with a cold block cache, so hit rates and read latency are worse until it warms. And the larger fleet raises the base rate of failures and the number of things to monitor. Practical consequences: scale before saturation (the plan above tolerates a peak of about 73% on 96 nodes, so growth at high utilization is risky), expand in small batches, watch P99 and compaction debt during the move, and keep a rollback of the topology change.
- Can a new node take traffic immediately? — It can once it holds replicas, but until its caches warm it is slower; some systems send it a share of reads gradually.
- How do you speed up a rebalance safely? — By raising the throttle in steps while watching foreground P99 and replication lag, never by removing the throttle.
How does tenant isolation work?
Isolation here has three layers and each answers a different failure. Identity and namespace: the tenant comes from a verified credential at the gateway, never from a request header the client can set, and it is prefixed to every key (atlas/…), applied in routing, storage, logs and admin operations. Otherwise Birch could read Atlas by crafting a key. Resource fairness: per-tenant quotas on request rate, bytes stored and in-flight requests, enforced at the front door with 429 responses, plus fair queuing inside the node so that one tenant's burst waits behind its own backlog rather than ahead of other tenants'. Atlas, the large hot tenant, can consume a share of a partition's capacity that others cannot take. Blast radius: placement choices decide who suffers when something goes wrong. A noisy tenant whose keys are hash-spread shares every partition with everyone, so a surge touches the whole fleet; a very large tenant can be pinned to a dedicated set of partitions or nodes so that its incidents stay inside that set, at the price of lower utilization. Encryption keys can be per tenant, so a key rotation or a revocation can be scoped to one customer. Audit logs and backups carry the tenant too, which is what makes a per-tenant restore or erasure possible. None of this is free: pinning wastes capacity, and quotas need tuning. The design decision to state is where the line sits between shared and dedicated resources.
- What if a tenant's single key is hot? — Quotas apply per tenant, and the key-level mitigations from the hot-key discussion apply inside that tenant.
- How do you stop a cross-tenant bug? — Make the tenant a mandatory argument in the storage API so no code path can build a key without one, and test it with a tenant-fuzzing suite.
Which metrics show that the store is healthy?
Metrics should answer whether the promises hold, not only whether machines are up. Latency and errors per operation and per read mode, at P50, P99 and P99.9, split by tenant, so a regression for Cedar is not averaged away by Atlas. Commit latency per partition (leader append to majority ack), since it exposes a slow disk or a slow follower. Replication health: under-replicated partitions (fewer than 3 healthy replicas), replication lag in entries and seconds for followers and for the remote region, leader elections per hour, and leadership churn. Storage engine: compaction debt in pending bytes, level-0 file counts, write stalls, disk-free against the 60% target, and Bloom-filter false-positive and block-cache hit rates. Data integrity: checksum failures, scrub results, repair age (time since each partition was last repaired, compared with the tombstone retention), and tombstone-to-live ratios. Routing: wrong-epoch rate, stuck migrations. Safety: count of 503s for lost quorum, 412 and 409 rates, idempotency replays. Resource isolation: throttle events per tenant and top-k hot keys. Alerts should be tied to user impact and to risk deadlines: an under-replicated partition is a warning, a second loss is an outage, so the paging threshold is based on how close the group is to losing its majority. A dashboard that is green while backups have not been restored in six months is lying about a promise.
- Which single metric would you look at first during an incident? — P99 latency and error rate by partition, because it localises the problem to one partition, one node or the whole fleet.
- How do you alert on repair age? — Compare it with the tombstone retention; if the oldest partition's last repair approaches 10 days under the assumed retention, page before resurrection becomes possible.
| Signal | What it tells you | Danger threshold (assumed examples) |
|---|---|---|
| Under-replicated partitions | Groups with fewer than three healthy replicas | Any group at two replicas for longer than the rebuild target; any group at one |
| Commit latency P99 per partition | Leader-to-majority acknowledgement time | Sustained rise over the latency objective, e.g. above 20 ms against a 20 ms assumption |
| Compaction debt and level-0 files | Background work falling behind ingest | Growing for several hours, or a write stall event |
| Disk-free | Headroom for compaction, snapshots, repair | Above the 60% fill target on any volume |
| Repair age | Time since partition replicas were compared | Approaching the tombstone retention (10 days in the example) |
| Replication lag to US | Distance of the asynchronous replica | Above the lag assumed in the RPO statement (300 ms example) |
| Wrong-epoch rate | Routers with stale maps | Spike that does not decay after a migration |
What are change streams for?
A change stream (Watch) exposes the ordered sequence of committed changes to keys so that other systems can follow along without polling. Meridian's reasons are concrete: invalidating near caches when a tenant's settings change, feeding a search or analytics index, triggering an audit trail, and replicating to the remote region. Because the replicated log already contains every committed change in order within a partition, a stream is a view over it: the client asks to start from a position (a log index or a version), receives entries as they commit, and can resume from the last position it processed after a disconnect. Guarantees matter more than the mechanism. Order is per partition, not global: two keys in different partitions have no common order, so consumers must not assume one. Delivery is at-least-once, because a consumer may crash after processing and before recording its position, so consumers must be idempotent, usually by keying on the key and version. The stream must also say what happens when the consumer falls behind past the log's retention: it receives a 'position too old' error and has to re-snapshot, which is why retention must be sized for consumer outages. Tombstones appear as delete events. It is an advanced, optional feature because it adds retention cost and a new class of slow-consumer failure, which should be isolated so it cannot slow the write path.
- Can the stream be linearizable? — Within one partition the order is the log order; consumers see changes after commit, with a delay.
- What if consumers are slow? — Their position is tracked per consumer; the write path is never blocked, and a consumer that falls beyond retention must resynchronise.
How do you test durability and failover?
By trying to break the promises while a checker watches, and by repeating it. The core technique is fault injection under load. Run a workload with many clients writing known values and recording every response, including timeouts, while a nemesis kills processes with SIGKILL, pauses them, partitions the network (one node, a leader, a minority, bridges), skews clocks, fills a disk and powers off machines in a test rig. After each run, a checker validates the history: every acknowledged write must be present, no read may return a value older than one already observed after an acknowledged newer write (the linearizability check, which is expensive and done on small histories per key), CAS must never allow two winners at one version, and no deleted key may return. Writes that timed out are allowed to have either outcome but never a mixed one. Jepsen-style testing is the best-known pattern for this, and the idea transfers even if you build the harness yourself. Specific durability tests: kill -9 right after acknowledgement and after fsync boundaries; truncate the log mid-record and verify recovery stops at the checksum failure; flip bits in an SSTable block and verify detection. Failover tests: kill leaders and measure the time to a new leader and the failed-request window; verify no acknowledged write is lost and that the old leader is fenced. Operational drills belong too: restore from backup into a scratch cluster, replace a node, run a game-day for a zone loss. Run a subset on every build, the long soak nightly, and publish results as evidence of the promises, not as a pass/fail badge.
- Can you prove it correct? — Formal models (such as TLA+ specifications of the protocol) catch design errors, and testing catches implementation errors; neither replaces the other.
- How do you test power loss without a power switch? — Simulate it by killing the process and by using a storage layer that drops unflushed writes, which is stricter than a plain process kill.
Regions and eventual consistency
Distance changes the economics of every guarantee. The three questions below separate what a user experiences, what physics allows, and what a recovery plan can honestly promise.
What does eventual consistency mean to a user?
It means that if updates stop, all copies converge to the same value, and it says nothing about what anyone sees in the meantime or how long convergence takes. For a person using Meridian that shows up as specific surprises. A tenant admin in Frankfurt saves a new checkout policy (v11) and a colleague in Virginia, reading from the US replica that is 300 ms behind, still sees v10 and may act on it. A user saves a setting, the page reloads from a different replica and shows the old value, as if the save failed, so they save again. Two readers see different values at the same moment and disagree in a meeting. A deleted item reappears for a moment. Convergence is usually fast, a few hundred milliseconds in this example, but during an incident it can stretch to minutes. Products live with this by choosing consistency per use. Preferences display, dashboards and caches can use eventual or follower-stale reads and be labeled. Anything that decides something, such as authorization or a feature gate that must flip for everyone at once, uses linearizable reads in the home region. read-your-writes with a version floor removes the most visible surprise, the vanishing own write, cheaply. Designers should also say what resolves a conflict when two regions accept writes concurrently, since 'eventual' without a conflict rule can converge on the wrong value; Bigtable's documentation, for example, describes server-time last-write-wins for conflicting cell writes across clusters, which is deterministic but can discard one of two updates.
- Is eventual consistency the same as 'wrong'? — No; it is a trade of recency for availability and latency, acceptable when the product tolerates staleness and has a clear conflict rule.
- How do you tell a user their view may be stale? — Return the read mode and version; UIs can show 'updated at' times, or request a read-your-writes token after saving.
Can you have strong consistency across continents?
Yes, and it costs latency on every write, so it is a deliberate product choice. To commit a write that survives the loss of a whole region, the leader must wait for acknowledgement from replicas in other regions before replying. If a round trip between Europe and the US east coast is, say, 80-100 ms (an assumed range, set by distance and routing, and it cannot be engineered away), then every write pays at least that, instead of the 1-2 ms of a replica group in one region. With three regions and majority commit, the leader's own region counts as one vote and the write needs one more, so it waits for the nearest other region's acknowledgement, and where the replicas are placed decides the cost. Reads can still be local if a follower proves it is current (read-index against the leader) but that again is a cross-region message unless the leader is local; so linearizable reads in a remote region also pay the distance or are served by the leader's region. Systems that offer this exist: multi-region strongly consistent modes in some managed databases, and consensus-based stores designed for it. Their details differ by product and by version, so verify current documentation rather than trusting a blog; for example, DynamoDB global tables are documented with both a multi-Region eventual-consistency mode and a separately configured strong-consistency mode, and the exact semantics should be read from the AWS documentation. The reference design here chooses the cheaper arrangement: one home region leader, asynchronous replication to US, with the US serving eventual reads, and it says honestly that async replication does not turn the pair into a single strongly consistent system. Pick global strong consistency only when the loss of acknowledged writes in a regional disaster is unacceptable and users can tolerate the write latency.
- Can a remote region serve linearizable reads in the reference design? — Only by routing them to the home region; the local replica can serve eventual or bounded-stale reads.
- What is the alternative to paying the latency? — Multi-active regions with conflict resolution, which keep local writes fast but change the contract to eventual consistency with merge rules.
What are RPO and RTO for multi-region, and what numbers can you promise?
RPO, recovery point objective, is how much acknowledged data you may lose, expressed as time or as a count of writes; RTO, recovery time objective, is how long the service may be unavailable. They are separate and each depends on mechanism. In the reference design the home region EU commits writes using its local group and ships the log to US asynchronously. Suppose the lag is 300 ms (assumed). If the EU region is lost at that instant, everything acknowledged in the last 300 ms may be missing in US. Assume EU receives 30% of the average 40,000 writes/s, which is 12,000 writes/s; in 0.3 s that is 12,000 × 0.3 = 3,600 acknowledged writes at risk. Concretely, atlas/feature=checkout_policy might be v11 in EU and v10 in US. So RPO is not zero; it is lag-dependent exposure, and the honest statement is 'up to roughly the replication lag at the moment of failure, with a bound only if lag is bounded and monitored'. A synchronous cross-region commit would make RPO zero for committed writes at the price discussed in the previous answer. RTO depends on detection time (health checks must distinguish a region outage from a partition, which is hard), the decision to fail over (automatic risks split-brain if the old region is merely isolated; a human-approved promotion is slower but safer), promotion of the US replica with a higher epoch so that the old region is fenced when it returns, DNS or routing changes with their caching, and warming. Examples such as 'RPO under 1 second, RTO 15 minutes' are targets to be tested in drills rather than promises. Residency complicates it further: EU-resident records may not be silently promoted in the US, so failover for them is either blocked or requires an explicit legal decision. After failover, writes accepted in US while old EU data was lost need reconciliation when EU returns.
- How does an old home region avoid split-brain when it comes back? — A higher epoch on the promoted region fences it; it must rejoin as a follower and discard or reconcile any unreplicated tail.
- Is a documented RPO of 'zero' ever honest with async replication? — No; zero requires that no acknowledgement precedes remote durability.
A complete spoken whiteboard explanation
What follows is roughly what a six-minute oral walkthrough could sound like once the questions above are absorbed. It is written to be spoken, with the board filling in as each problem appears.
"I will treat this as a system of record for tenant settings: two billion records of about a kilobyte, so about two terabytes of logical data, mostly reads, and writes that must not be lost once acknowledged. My assumptions are on the board and I will say when one of them drives a decision."
"Start with one machine. A hash map in memory is fast and loses everything when the process dies, so before replying to a write I append it to a write-ahead log and fsync, then apply it. Reads come from a sorted in-memory table backed by immutable sorted files, with Bloom filters so a missing key does not cost a disk read, and compaction merges the files in the background. That is an LSM engine, chosen because a sustained stream of writes becomes cheap sequential I/O, with Bloom filters and a cache keeping reads acceptable."
"One machine cannot serve a million reads per second at peak, so I split the keyspace into 1,024 fixed virtual partitions of about 2 GB each. A router hashes the tenant-prefixed key to a partition and consults a small partition map, whose entries carry an epoch so a stale router gets a redirect rather than a wrong answer. Modulo hashing would move 75% of the data when I add a node; fixed partitions move only the share the new node should own, about a quarter."
"One copy per partition is a single point of loss, so each partition is three replicas on separate failure domains with one leader. A write becomes committed when a majority, two of three, have it durably in their logs, and only then is it acknowledged. Terms fence old leaders; a minority partition refuses writes with an explicit unavailable error, because I would rather reject than return a false success. Two simultaneous replica failures stop commits for that partition; replication factor three tolerates one."
"Reads default to linearizable through the leader with a read-index check; I offer labeled follower-stale reads and read-your-writes through a version floor for latency. Conditional writes are commands in the same log, so two admins editing at version 7 get one success and one 412, and retries carry idempotency keys so an ambiguous timeout cannot double-apply."
"Deletes write tombstones kept longer than the longest time a replica can be away; TTLs are replicated deletes decided on the leader's clock. Backups are snapshots plus shipped log so I can restore to a point in time, and I rehearse restores. For hot keys I use labeled follower reads and caches, and I say writes to one key stay serialized at one leader. For geography, the home region is EU with an asynchronous US replica about 300 milliseconds behind in my example, so the recovery point is not zero, and failing over needs a higher epoch to fence the old region. I would watch under-replicated partitions, commit latency, compaction debt, repair age and replication lag, and I would test it by killing leaders under load and checking that no acknowledged write disappears."
How a strong answer differs from a shallow one
A rubric helps because box-drawing and reasoning can look similar for the first ten minutes. The columns below describe what each level of answer tends to contain; they are a teaching aid, not any company's hiring standard.
| Dimension | Shallow | Solid | Deep |
|---|---|---|---|
| Durability | Says 'we replicate, so it is safe'. | Explains write-ahead log, fsync before acknowledgement and replay. | Separates page cache from stable storage, handles torn tails by checksum, ties acknowledgement to majority commit and notes volatile-cache disks. |
| Partitioning | Hash mod N. | Consistent hashing or virtual partitions, with a partition map. | Computes the 75% vs 25% movement, uses epochs to fence stale routers, and describes a learner-based safe move with throttling. |
| Replication | 'Write to three nodes.' | Leader, majority commit, RF=3 tolerates one failure. | Explains why RF is not consensus, terms and fencing, election overlap, and that two failures stop commits. |
| Consistency | 'Strong' or 'eventual' as slogans. | Names linearizable and stale reads and when each is used. | States the scope of linearizability, read-index or lease cost, and why R+W>N is not linearizable. |
| Concurrency | Read, check, write. | Compare-and-set with a version. | Places the check in the replicated state machine, handles 412, retries and idempotency keys with retention. |
| Failure honesty | No failure story or false success on partition. | Describes leader failure and re-election. | Walks the minority-partition timeline, ambiguous timeouts, resurrection and RPO exposure with numbers. |
| Numbers | None, or unlabelled. | Reasonable capacity estimate. | Derives per-node load from labeled assumptions, distinguishes throughput-bound from capacity-bound, and says what would change the plan. |
| Operations | Says 'add monitoring'. | Lists metrics. | Ties alerts to risk deadlines (repair age vs tombstone retention), covers restore drills and fault-injection testing. |
Common candidate errors and the corrections
Each item pairs a statement that sounds plausible with the correction, because the correction is the thing worth remembering.
- "With replication factor 3, any two machines can fail and writes still succeed." Correction: a majority of three is two, so one failure is tolerated; with two lost, commits stop and strongly consistent reads are unavailable.
- "R plus W greater than N gives linearizability." Correction: it guarantees that quorums overlap; sloppy quorums, concurrent writes, clock skew, partial writes and membership changes mean it is not linearizability.
- "We write to the log, so it is durable." Correction: a write that only reaches the OS page cache can vanish on power loss; durable means flushed with fsync (on a stack that honours flushes) and replicated per the commit rule.
- "Use modulo hashing; it is simple." Correction: changing the node count remaps about 75% of keys when going from 3 to 4 nodes; use fixed virtual partitions or consistent hashing.
- "Read the value, check the version in the app, then write." Correction: that is check-then-put and races; the comparison must be part of the replicated write.
- "A timeout means it failed, so retry." Correction: the outcome is unknown; retry with the same idempotency key.
- "Delete removes the data." Correction: delete writes a versioned tombstone; physical removal follows compaction and backup retention, and dropping the tombstone too early resurrects values.
- "Async cross-region replication gives us global strong consistency." Correction: it gives a recovery copy with a non-zero RPO; linearizable reads are served only in the home region.
- "Add nodes to fix the hot key." Correction: a single key lives in one partition and its writes go through one leader; use labeled follower reads, caches and coalescing for reads.
- "The cache is the database." Correction: a store that may evict or lose data silently is a cache; a system of record needs the durability contract.
- Say the guarantee's scope and its failure condition in the same sentence.
- Mark every number as an assumption and show the arithmetic.
- Name the read mode on each read path and say what it can return.
- Describe what the system does when it cannot decide safely: reject with an explicit error.
- Mention how you would test the claim, not only how you would build the feature.
Pushbacks and curveballs
Interviewers often interrupt a clean design with a twist. These are twists with defensible responses, in the form of a lookup rather than a script.
| Curveball | A defensible response |
|---|---|
| Your leader's disk is slow but not dead. Writes crawl. | Commit latency rises for that partition because the leader waits on its own fsync. Detect it from commit-latency and disk metrics, transfer leadership to a healthy replica, and treat the slow disk as degraded rather than waiting for it to fail outright. |
| Two of three replicas of one partition are lost in the same rack failure. | Commits for that partition stop and strongly consistent reads fail, with 503, since no majority exists. Recover by restoring replicas from the survivor or backup; this is why replicas span failure domains. Forcing the lone survivor to lead would risk losing writes that were committed on the lost pair, so it needs an explicit operator decision. |
| A customer needs a transaction across two keys. | Out of scope for the reference store. Options: put both fields in one record so the single-key CAS covers them, use a transactional store for that data, or build a cross-partition protocol such as two-phase commit with its blocking and recovery costs. |
| Can we make it faster by acknowledging after the leader's fsync only? | Yes, and writes become faster, but a leader loss before replication can lose an acknowledged write. That is a different durability promise and must be a named, opt-in mode. |
| Traffic doubles overnight. | Check which resource saturates (the plan above uses about 73% of a node at short peak), shed or throttle by tenant, add nodes in small batches with rate-limited moves, and warn that moves themselves add load. |
| The partition map service is down. | Routers use cached maps and nodes still check epochs, so data-plane traffic continues; ownership changes, splits and failover of partitions wait until the map service is back. |
| A bug writes bad values for an hour. | Replication copied it everywhere. Restore from a snapshot plus the log up to just before the bad deploy into a scratch cluster, then copy the affected keys back, and reapply any legitimate writes and deletes since. |
| The interviewer insists on a leaderless design. | Switch cleanly: N=3, quorum reads and writes, hinted handoff, read repair and anti-entropy, with versions or last-write-wins for conflicts. State that the guarantees are eventual with tunable recency, and keep conditional writes as an explicit exception that needs its own mechanism. |