Engineering Atlas › System Design Interviews

Design a distributed key-value store

20 min read · last reviewed 2026-10-17

Open in the interactive app: practice, quiz, mock interview →

In one minute

Requirements

Functional

Non-functional

Out of scope: multi-key transactions, secondary indexes, range scans and SQL.

Back-of-envelope

API

service KV {
  rpc Get(GetRequest) returns (GetResponse);
  rpc Put(PutRequest) returns (PutResponse);
  rpc Delete(DeleteRequest) returns (PutResponse);
}
message GetRequest  { bytes key = 1; Consistency read = 2; }        // ONE | QUORUM | ALL
message GetResponse { repeated Versioned values = 1; bytes context = 2; } // >1 value = siblings
message PutRequest  { bytes key = 1; bytes value = 2; bytes context = 3;
                      Consistency write = 4; uint32 ttl_s = 5; }
message PutResponse { bytes context = 1; }
message Versioned   { bytes value = 1; bytes version = 2; bool tombstone = 3; }

context is an opaque version that the client got from a read. When the client passes it back on a put, it says "this write replaces what I read". That is how the store tells an overwrite apart from a concurrent write.

Data model

High-level design

flowchart LR
  CL["Client (ring-aware)"] -->|"put k"| CO["Coordinator node"]
  CO -->|"replicate"| A["Replica A"]
  CO -->|"replicate"| B["Replica B"]
  CO -.->|"down: hint stored"| H["Hint queue"]
  H -.->|"on recovery"| C["Replica C"]
  A --> W1["WAL + memtable"]
  W1 --> S1["SSTables + Bloom"]
  A <-->|"gossip, Merkle sync"| B
  B <-->|"gossip, Merkle sync"| C

Write path.

  1. The client hashes the key and sends the request to a coordinator, usually the first healthy node in the key's preference list.
  2. The coordinator stamps a version and sends Put to all N replicas.
  3. Each replica appends to its WAL, which group commits and then fsyncs. It inserts into the memtable and acknowledges.
  4. After W acknowledgements, the coordinator returns success.
  5. If a replica is down, the coordinator writes to the next healthy node on the ring instead (a sloppy quorum). That node stores a hint naming the intended owner and hands the data back when the owner recovers.

Read path.

  1. The coordinator asks all N replicas for the key. It can ask for a full value from one replica and digests from the others, to save bandwidth.
  2. It waits for R responses.
  3. It returns the newest version, or several siblings if versions are concurrent.
  4. Stale replicas get the newest version asynchronously. This is read repair.

On each replica, a read checks the memtable first, then SSTables from newest to oldest. A Bloom filter check skips the SSTables that cannot contain the key.

Deep dives

1. Partitioning and rebalancing

Hash mod N moves almost every key when N changes, which is unworkable.

Consistent hashing with one token per node moves only the keys that neighbour the changed node. But load is uneven, with arcs differing by several times, and a new node takes load from only one neighbour.

Virtual nodes give each node many tokens. Its share of the ring is then the sum of many small arcs. With purely random tokens you need about 256 per node for that to average out, which was Cassandra's old default; a token allocator that picks positions to balance load gets there with 16, and Cassandra's docs warn that more tokens raise the chance that a double failure causes an outage (Cassandra docs). A joining node pulls small ranges from many peers in parallel. Heterogeneous hardware is handled by giving bigger machines more tokens.

Fixed partitions split the hash space into, say, 4,096 ranges up front and assign ranges to nodes through a metadata service. This makes moving data a clean "move partition p" operation. Dynamo itself ended up here: its paper compares three strategies and moved production to equal-sized partitions with a fixed number per node, because random tokens made bootstrapping and archiving slow (Dynamo paper).

I chose virtual nodes for a Dynamo-style peer system, and I'd switch to fixed partitions if we already ran a consensus-backed metadata store. Both beat one token per node. The cost of vnodes is that a node participates in many replica groups, which widens the set of keys affected when any two nodes fail together.

2. Quorums, sloppy quorums and repair

With N=3, the common settings are below. Dynamo's most common production setting was (N, R, W) = (3, 2, 2) (Dynamo paper).

What overlap does and does not give you. R + W > N guarantees that a read sees at least one replica holding the latest acknowledged write. It is still not linearizable:

Convergence takes three mechanisms:

Choice: QUORUM/QUORUM by default, hinted handoff on, and anti-entropy running continuously with a throttle. Know the alternative too: Amazon's managed successor, DynamoDB, dropped leaderless quorums. Each partition is a Multi-Paxos group whose leader holds a lease and takes all writes and strongly consistent reads, and its replicas store data in a B-tree rather than an LSM tree (DynamoDB paper).

3. Versioning and conflicts

Last-writer-wins on timestamps is simple and gives one value per key. But clock skew silently discards writes: a node 50 ms fast beats a later write from a correct node. Hybrid logical clocks reduce the damage but cannot remove it, because two genuinely concurrent writes still have to lose one.

Vector clocks (or dotted version vectors) record which coordinator has seen which counter. If one version dominates the other, it wins. If neither does, the store keeps both as siblings and returns both on the next read, and the client merges them and writes back with the context. Nothing is lost. The costs are client complexity and version metadata that grows with the number of coordinators. Dynamo dropped the oldest entry once a clock reached a threshold of about 10 pairs. Conflicts are rarer than this machinery suggests: over 24 hours, 99.94% of requests from Amazon's shopping cart service saw exactly one version (Dynamo paper).

CRDT values merge automatically, but only for counters, sets and maps.

Choice: LWW with hybrid timestamps by default, since most keys are cache-like or single-writer. Siblings with dotted version vectors are an opt-in per table for data where losing a write is a bug, such as shopping carts.

Deletes write tombstones that must outlive the longest possible outage plus the repair interval, typically about 10 days (Cassandra's gc_grace_seconds default of 864,000 seconds, per the CQL docs). If a tombstone is purged too early, a replica that missed the delete can bring the value back.

Variant: distributed cache

A cache in front of a database inverts most of the choices above, because the cache is not the source of truth. The requirement that drives everything is the hit rate: at 95% the database sees 5% of reads, at 80% it sees four times that, which is often an outage (AWS Builders' Library). Targets shift accordingly, to sub-millisecond p99 reads, millions of reads per second, and values of 100 B to 10 KB, with durability optional and a restart allowed to lose data.

Name which of the two systems you are building in the first minute of an interview. A cache that quietly gains replicas and quorums is a slow database, and a store that quietly gains eviction loses data.

Down the stack

Layer 1The contract

  • A put acknowledged at W is in the WAL of at least W replicas, or of stand-in nodes holding hints for them.
  • A get at R returns every version that is not dominated by another, among the R replicas that answered.
  • With R + W > N, no sloppy substitutions and no concurrent writes, a get observes every write acknowledged before it began.
  • Without new writes, all N replicas converge within the anti-entropy period, for example 24 hours, and usually within seconds through read repair and hints.
  • Latency: p99 under 10 ms at QUORUM within one availability zone set.

Layer 2The mechanism

Messages. Coordinators send MutationRequest{key, value, version, ttl} and replicas answer MutationAck. Reads send ReadRequest{key, digest_only} and get back either a full value or a hash.

Failure detection. Gossip runs every second. Each node exchanges heartbeat generations and versions with a few peers. A phi-accrual failure detector turns heartbeat arrival gaps into a suspicion level instead of making a binary timeout decision. Cassandra works this way (Cassandra docs).

Storage engine. Writes append to a WAL segment (Cassandra's commit log uses 32 MiB segments by default, per the Cassandra config) and go into a memtable. When the memtable fills (say 64 to 256 MB), it is flushed as an SSTable, which holds sorted data blocks of about 4 to 64 KB, a sparse index, a Bloom filter at about 10 bits per key (roughly 1% false positives, per the RocksDB wiki) and a footer. Compaction merges SSTables, drops overwritten versions and expired tombstones, and keeps the number of files a read must check small.

Layer 3The machine

  • Write cost. The WAL append is a sequential write to the page cache followed by fsync. On NVMe with power-loss protection, fsync takes about 20 to 100 µs, and group commit shares one flush among many writes. A replica write therefore costs microseconds of device time, and the network round trip (about 200 µs within an AZ, 1 to 2 ms across AZs) dominates the latency of a W=2 write.
  • Read cost. A read that misses the page cache costs one random 4 KB NVMe read, about 80 to 100 µs. Bloom filters and block indexes held in RAM keep that to about one device read per lookup. On spinning disks the same read would take about 5 to 10 ms, which is why this design assumes SSDs.
  • Compaction. It rewrites data many times. Write amplification for leveled compaction is often above 10× (RocksDB wiki) and 10 to 30× is a fair planning range, so a node absorbing 6k writes/s × 1 KB = 6 MB/s of user data may write 60 to 180 MB/s to disk. That is why compaction is throttled, and why SSD wear is a cost to plan for.
  • Memory. In JVM-based stores a 200 ms GC pause makes a node look dead to gossip, one reason ScyllaDB rebuilt Cassandra's design in C++.

Why it's built this way

  1. Machines and disks fail daily at 100-node scale, so every key needs N replicas on independent failure domains.
  2. Synchronous agreement among all replicas would block writes during failures. Quorums let a majority, or W, proceed instead.
  3. Proceeding without everyone means replicas diverge. That makes versions and repair mechanisms mandatory, not optional.
  4. Random writes are slow on every device, and LSM trees turn writes into sequential appends. Their cost on reads is paid back with Bloom filters and compaction.
  5. Moving data costs network and disk time, so consistent hashing with vnodes keeps movement proportional to the change and spreads it across peers.

Failure modes

In the wild

Trade-offs

OptionStrengthsWeaknessesChoose when
Leaderless quorum (Dynamo)Writes available during failures, no failoverConflicts, repair machinery, not linearizableCarts, sessions, profiles, high write availability
Leader per partition (Raft)Linearizable, no conflictsElection pause, writes need a majorityMetadata, money, locks
LWW timestampsSimple, one valueSkew silently loses writesCache-like or single-writer data
Vector clocks / siblingsNo lost writesClient merge logic, metadata growthMulti-writer data where loss is a bug
vnodes on a ringEven load, parallel rebuildMore replica sets exposed to double failuresPeer-to-peer membership
Fixed partitions + metadata serviceClean moves, simple reasoningNeeds a consistent metadata storeYou already run etcd or ZooKeeper
LSM engineFast writes, compressionRead and space amplification, compaction I/OWrite-heavy workloads

Leadership lens

Check yourself

  1. A deployment uses N=3 with R=2, W=2, so R + W > N. Why is a read still not guaranteed to be linearizable?

    Show answer

    Overlap only guarantees that a read sees at least one replica holding the latest acknowledged write. A write still in flight can be visible to one reader and not the next, and sloppy quorums write to stand-in nodes outside the preference list, which breaks the overlap until hints are delivered.

  2. When would you pick vector clocks with siblings over last-writer-wins, and what does each choice cost?

    Show answer

    LWW is simple and returns one value, but clock skew silently discards writes, so it suits cache-like or single-writer data. Siblings with dotted version vectors lose nothing, which matters for multi-writer data like shopping carts, at the cost of client merge logic and growing version metadata.

  3. Why does the lesson estimate about 100 nodes today, and why does storage rather than throughput size the cluster by year 3?

    Show answer

    Replica work is about 3M reads/s plus 600k writes/s, and at roughly 50k operations per node that is about 72 nodes, or about 100 with 30% headroom. Those 100 nodes hold only about 300 GB each today, but with 300 GB/day of new raw data replicated three times, storage reaches around 4 TB per node by year 3 and becomes the binding constraint.

  4. What goes wrong if tombstones are purged too early, and how long should they be kept?

    Show answer

    A replica that missed the delete still holds the old value, and once the tombstone is gone repair can bring the deleted value back. Tombstones must outlive the longest possible outage plus the repair interval, typically about 10 days, as with Cassandra's gc_grace_seconds default.

Sources and further study

Primary

Systems docs

Books

Other curricula

Watch

Free resources

Open in the interactive app: practice, quiz, mock interview →