The problem
You’re asked to design a distributed key-value store — the system underneath
DynamoDB, Cassandra,
or Riak. The interviewer says something like: “Design a storage
system that supports get(key) and put(key, value) across many machines. It should
scale horizontally and survive machine failures.”
Sounds like a big hash map. It isn’t. The trap is treating partitioning as the whole question: candidates draw a hash ring, say “consistent hashing,” and stop. The question is really about what happens when machines fail while reads and writes are in flight — and at the scale this system runs, machines are always failing. Replication, quorum arithmetic, and conflict resolution are where the design actually lives.
Important
Key takeaway: A distributed KV store forces you to choose what happens during a network partition before you design anything else. Stay available and accept conflicting writes (Dynamo), or stay consistent and refuse some requests (a consensus-based store)? Every other decision — partitioning, replication, the API itself — follows from that contract. This walkthrough builds the always-writable (AP) version, because that’s what “Dynamo-style” means.
Below is how I’d walk through this, start to finish.
1. Requirements
First 3–5 minutes. Questions I’d ask to pin down what we’re building:
Functional requirements
- R1.
put(key, value)— write a value by key. - R2.
get(key)— read a value by key. - R3.
delete(key)— remove a key. - R4. Scale horizontally — add and remove nodes without downtime or manual resharding.
- R5. Durability — an acknowledged write survives the permanent loss of a node.
Scope
This design covers the distributed coordination layer. Explicitly out of scope:
- Range queries and secondary indexes (a different storage-engine question — they constrain partitioning in ways point lookups don’t)
- Cross-record transactions (a consensus/2PC question; see distributed transactions)
- Multi-region active-active (named in the evolution table; a design of its own)
- Authentication, multi-tenancy, quotas (real products need them; not design drivers here)
- Storage-engine internals (LSM compaction tuning is a deep topic; I name the engine and move on)
Non-functional requirements
- Scale. How many keys, and how large are values? Total dataset size determines fleet size, and fleet size determines how often we’re handling failures.
- Read/write mix. KV stores are often closer to balanced (70/30) than the 100:1 of a read-heavy web app — caching helps less than usual, so the storage path itself must be fast.
- Latency. Single-digit-millisecond p99 for both operations?
- Availability vs consistency. During a partition, must writes always succeed? This is the contract question — it decides the architecture.
- Durability. Replication factor, and whether acknowledged writes may ever be lost.
Say the interviewer confirms: 100B keys, values average 1 KB (max 100 KB), 1M reads/sec and 300K writes/sec at peak, p99 < 10 ms, replication factor 3, single region to start — and the hard one: writes must succeed even during partial failures. Always-writable.
Note
Interview signal: Asking “what should a read return immediately after a write during a network partition?” is the question that separates strong candidates. It forces the interviewer to choose between availability and consistency — and whichever they choose, you now know which half of the CAP space you’re designing in. Candidates who never ask end up designing a CP system for an AP prompt, or vice versa.
2. Capacity estimate
- Storage: 100B keys × 1 KB ≈ 100 TB raw. With replication factor 3 plus storage-engine overhead, ~350–400 TB on disk.
- At ~2.5 TB of usable data per node (leaving headroom for compaction and repair), that’s a ~150-node fleet.
- Throughput per node: 1M reads/sec ÷ 150 ≈ 7K reads/sec/node — comfortable for an SSD-backed store. The fleet is storage-bound, not CPU-bound.
Two conclusions worth saying out loud:
“First — at 150 nodes, hardware failure is not an event, it’s a steady state. Between disk failures, kernel patches, and network blips, something in this fleet is down or degraded every few days. Failure handling can’t involve an operator; it has to be automatic. Second — since the fleet is sized by storage, the expensive operation is moving data, not serving requests. Rebalancing and repair traffic are first-class design concerns, not afterthoughts.”
3. High-level architecture
Three logical subsystems:
- Request coordination — receives a
get/put, finds the replicas for the key, fans out, collects acknowledgments. In a leaderless design, any node can coordinate any request. - Partitioning and membership — decides which nodes own which keys (the hash ring) and tracks which nodes are alive (gossip). This is what makes R4 (elastic scaling) possible.
- Replication and repair — keeps N copies of every key in sync through quorums, hinted handoff, and background anti-entropy. This is what makes R5 (durability) possible.
flowchart TD
Client[Client] --> Router[Routing layer]
Router --> Coord[Coordinator · node A]
Coord --> StoreA[(Storage · A)]
Coord -->|replicate| NodeB[Node B]
Coord -->|replicate| NodeC[Node C]
classDef store fill:#eef2f1,stroke:#2c5f5d,stroke-width:2px,color:#1f2937;
class StoreA store
Membership decides the system’s character. A store with a central metadata master and a store where nodes gossip peer-to-peer are different systems with different failure modes — and this question is asking for the second one.
4. Architecture evolution
A key-value store in production evolves through recognizable levels:
| Level | Architecture | Trigger for next level |
|---|---|---|
| 1. Single node | One server: hash map + write-ahead log | Dataset exceeds one machine; SPOF |
| 2. Primary/replica | One writer, read replicas, failover | Write throughput ceiling; failover drops writes |
| 3. Sharded primaries | Hash/range shards, each with a primary | Resharding pain; per-shard failover still blocks writes |
| 4. Leaderless ring | Consistent hashing, quorum R/W, any node coordinates | Multi-region latency; conflict semantics outgrow vector clocks |
| 5. Multi-region / CRDTs | Active-active regions, CRDT value types | (Production system; beyond interview scope) |
Each level preserves the previous one as a fallback in spirit — a Level 4 ring degrades to “fewer healthy replicas, same protocol,” which is precisely its advantage over Level 3, where a primary’s death is a special case requiring election.
The design in this walkthrough is Level 4: a leaderless ring with tunable quorums — the Dynamo paper architecture that Cassandra and Riak industrialized.
I’d say out loud: “I’m designing for Level 4 — leaderless with quorums. The reason to prefer it over sharded primaries is that node failure stops being a special case: there’s no election, no failover window, just a quorum assembled from whoever’s alive.”
Target architecture (Level 4)
This is the complete system. Every section that follows explains one part of this diagram:
flowchart TD
Client[Client] --> Router[Routing layer]
Router --> Coord[Coordinator · node A]
Coord --> StoreA[(Storage · A)]
Coord -->|replicate| NodeB[Node B]
Coord -->|replicate| NodeC[Node C]
NodeB --> StoreB[(Storage · B)]
NodeC --> StoreC[(Storage · C)]
Coord --> Hints[(Hint buffer · A)]
Coord <-.->|gossip| NodeB
NodeB <-.->|gossip| NodeC
StoreB <-.->|merkle repair| StoreC
classDef store fill:#eef2f1,stroke:#2c5f5d,stroke-width:2px,color:#1f2937;
class StoreA,StoreB,StoreC,Hints store
5. Core design choice: who accepts writes
Important
Key takeaway: The central question in a distributed KV store is where a write is allowed to land. One node per key (primary), a consensus group, or any replica? This choice sets the availability contract, the latency floor, and whether conflicts can exist at all.
Note
The AP/CP fork: This question and the distributed database walkthrough are the two branches of the same CAP decision. This walkthrough builds the AP side — always-writable, leaderless, tunable consistency. The database walkthrough builds the CP side — consensus per range, zero data loss, serializable isolation. Every mechanism in each design exists because of which branch was chosen first.
This is the question. Three options:
(a) Primary per partition. Each key range has one primary that orders all writes; replicas follow. Reads can be served from replicas (stale) or the primary (consistent). On primary failure, a failover process promotes a replica — and writes to that range block until it finishes.
(b) Consensus group per partition. Each key range is a Raft group; every write commits through a majority. Strong consistency, automatic failover — but every write pays a round of consensus, and during a partition the minority side refuses writes. This is a CP store (etcd, CockroachDB’s KV layer).
(c) Leaderless with quorums. Any of the N replicas accepts a write; the coordinator declares success after W acknowledgments. Reads ask R replicas and reconcile. No failover process exists because there’s nothing to fail over — but two replicas can accept conflicting writes during a partition, so the system must detect and resolve conflicts.
Why leaderless wins here
The justification is contract-specific: the interviewer said writes must succeed during partial failures. Options (a) and (b) both have a window — failover or minority partition — where some writes block. Option (c) has no such window by construction: as long as any W replicas (or stand-ins; see hinted handoff) are reachable, the write succeeds. The cost is real and named up front: we are choosing to handle conflicting versions rather than prevent them. Dynamo made exactly this trade for Amazon’s shopping cart — for that product, refusing a write (losing a sale) is strictly worse than merging two cart versions later.
Warning
Production reality:
“Leaderless” doesn’t mean “consistency-free.” Real deployments tune per-operation: a shopping cart writes at W=1 for availability, while a payment-adjacent read might use R=N for safety. The architecture’s actual product is the dial, not a fixed setting. DynamoDB exposes this as eventually-consistent vs strongly-consistent reads; Cassandra exposes it per query as ONE/QUORUM/ALL.
Deep contrast: Consensus group vs Leaderless quorum
This contrast is the Staff-level test on this question. Knowing both designs and their specific failure-mode differences is the signal.
| Dimension | (b) Consensus group (Raft/TiKV) | (c) Leaderless quorum (Dynamo/Cassandra) |
|---|---|---|
| Consistency | Strong (CP) — linearizable by default | Tunable (AP) — eventual by default, strong at R=N cost |
| Leader role | One elected leader per partition range; all writes go through it | No leader — any node coordinates any request |
| Write path | Client → leader → Raft majority quorum → commit → apply → ack | Client → coordinator → parallel fan-out to W replicas → ack |
| Write latency | Majority round-trip on every write (1 extra RTT for consensus log) | W parallel acks only (no ordering round) |
| Network partition | Minority partition refuses writes — no leader, no quorum | Both partitions accept writes — sloppy quorum with stand-ins |
| Conflicts | Impossible — consensus orders all writes | Possible — must detect (vector clocks) and resolve (client merge) |
| Failover | Raft election on leader death (~seconds) — writes block during election | None — coordinators are interchangeable, no election exists |
| Read repair | Not needed — leader is always authoritative | Required — replicas diverge; reads detect and heal staleness |
| Best for | Financial data, inventory, metadata stores (etcd, TiKV, CockroachDB KV) | User state, session data, high-scale eventually-consistent stores (Dynamo, Cassandra, Riak) |
The partition decision in concrete terms:
Consensus group under partition: if 3-node Raft group loses 2 nodes, the remaining node rejects all writes. Zero data loss, zero availability. The minority side is safe to read from (strong consistency) but refuses writes.
Leaderless under partition: each side keeps accepting writes into its available replicas with hinted handoff. When the partition heals, conflicting versions surface as siblings and get reconciled. Data is never lost; the price is reconciliation work and temporary divergence.
Tradeoff matrix
| Dimension | (a) Primary per partition | (b) Consensus group | (c) Leaderless quorum |
|---|---|---|---|
| Write availability during failure | Blocks during failover (seconds) | Minority side refuses writes | Continues if any W stand-ins reachable |
| Read consistency | Strong from primary; stale from replicas | Strong (linearizable) | Tunable via R+W vs N |
| Write latency | 1 node + async replication | Majority round-trip on every write | W parallel acks (no ordering round) |
| Conflicts | None (single writer) | None (consensus orders) | Possible — must detect and resolve |
| Operational complexity | Failover tooling, primary placement | Consensus tuning, group rebalancing | Conflict semantics, repair pipelines |
Architecture decisions
| Decision | Chosen | Rejected | Rationale |
|---|---|---|---|
| Coordination model | Leaderless quorum | Primary/partition, Raft groups | Always-writable contract; no failover window exists to manage |
| Partitioning | Consistent hashing with virtual nodes | Range partitioning | Point lookups only (no range scans in scope); uniform key distribution; minimal data movement on topology change |
| Membership | Gossip protocol | Central config service (ZooKeeper) | A membership master reintroduces the SPOF the leaderless design just removed |
| Conflict resolution | Vector clocks + client merge | Last-write-wins timestamps | LWW silently drops concurrent writes; clock skew makes “last” a lie |
| Failure patch | Sloppy quorum + hinted handoff | Strict quorum (fail the write) | Strict quorum violates the always-writable contract on a single node failure |
Core data structures
| Structure | Lives where | Role |
|---|---|---|
| Hash ring | Every node + smart clients (via gossip) | Maps hash(key) → preference list of N replicas |
| Version vector | Stored with every value | Detects concurrent writes vs causally-ordered ones |
| Hint buffer | Coordinator-local | Holds writes owed to temporarily-dead replicas |
| Merkle tree | Per node, per key range | Cheap diff for background anti-entropy repair |
Full schemas and access patterns are in §10. What matters now: all four structures exist to answer one question — which replicas have the latest version of this key, and how do we know?
6. Algorithm deep-dive: consistent hashing and cluster scaling [R4]
The ring mechanics
- Hash the entire key space onto a ring — say the range
0to2^128 − 1, wrapping around. Consistent hashing means both keys and nodes are hashed onto the same ring. - Each node owns one or more tokens (positions on the ring). A key belongs to the
first node found walking clockwise from
hash(key). - The preference list for a key is the next N distinct physical nodes clockwise.
With N=3, key
Kat position0x6A…might map to nodes B, C, D. “Distinct physical” matters once virtual nodes exist — two adjacent tokens can belong to the same machine, and replicating to yourself twice is not replication.
Tip
Why this beats mod n:
With hash(key) mod 150, adding node 151 changes the placement of ~99% of all keys — a 350 TB shuffle to add one machine. With consistent hashing, the same operation moves ~1/k of the data (where k is cluster size). For a storage-bound fleet, this isn’t an optimization; it’s the difference between “we add nodes on Tuesday afternoons” and “we schedule a maintenance quarter.”
Scale-out: adding a node
When node E joins with one token at position 0x80 in a four-node ring (tokens at 0x20, 0x60, 0xA0, 0xE0):
Before: 0x60 → [node C owns range (0x60, 0xA0)]
After: 0x60 → [node E owns (0x60, 0x80]], [node C owns (0x80, 0xA0]]
Node C streams only the range (0x60, 0x80] to E. Every other node is unaffected. Total data moved ≈ 1/k of total dataset.
The single-token problem: with one token per node, the entire transferred range comes from one source node (C). C’s I/O doubles during bootstrap while the other 149 nodes are idle. At 350 TB total, that’s a ~2.3 TB sequential stream from a single machine — hours of elevated latency, all concentrated on one node.
Virtual nodes (vnodes): distributing scaling pressure
Give each physical node ~256 virtual tokens scattered uniformly around the ring. When node E joins with 256 tokens:
- Each token steals a small slice from a different existing node
- Data streams to E from ~256 different source nodes in parallel
- Each source node gives up only ~(1/256) × (1/k) of its data — negligible I/O impact per node
- E’s bootstrap completes in a fraction of the time: 256 parallel 10 GB streams instead of one 2.3 TB stream
The same logic applies on scale-in (node departure or failure):
Single token: one clockwise successor absorbs 100% of the departed node's range → load spike
256 v-nodes: 256 clockwise successors each absorb ~(1/256) of the departed range → flat
With vnodes, failure recovery is embarrassingly parallel. At our 150-node fleet — where §2 said data movement is the expensive operation — this is the difference between a 1-hour and a 24-hour repair window.
The vnode tradeoff to name: more vnodes = more uniform load but more gossip metadata (each node tracks 150 × 256 = 38,400 token entries) and more repair coordination messages. Production systems typically use 64–256 vnodes per physical node; Cassandra defaults to 16 in newer versions after finding 256 was metadata-heavy for large clusters.
Scale-in: node departure and failure
Planned departure (decommission): the leaving node announces its exit via gossip, streams its ranges to successors, waits for transfer acknowledgment, then removes its tokens. Reads and writes to its ranges are served normally during transfer — coordinators route to the leaving node or its new owner based on which version is fresher.
Unplanned failure: the node disappears. Gossip marks it suspect, then down after a timeout. Its successor nodes absorb the key ranges immediately — sloppy quorum + hinted handoff covers short outages (§7); Merkle anti-entropy re-replicates from other RF copies for permanent loss (§9).
Gossip: broadcasting topology changes
Ring membership is not stored in a central master — that would reintroduce the SPOF the leaderless design removed. Instead, every node maintains its own view and reconciles via gossip:
- Every second, each node selects a few random peers and exchanges its full ring state (node IDs, token positions, heartbeat counters, node status).
- Information spreads epidemically: after O(log N) rounds (~7 rounds in a 150-node cluster), every node has seen the update.
- Generation + version counter on each node’s ring entry: a higher generation means the node restarted; a higher version means its state changed within a generation. Newer always wins.
A topology change (node join/leave) propagates to the full cluster within seconds, not minutes. A node that sees a peer as suspect starts routing around it speculatively — this is a local suspicion, not a quorum-agreed fact, which is exactly right: in this design, a false suspicion means a few unnecessary hints, not a failover.
Seed nodes are a bootstrapping mechanism: new nodes need to find at least one existing cluster member to join. Seeds are well-known addresses (typically 2–3 per region) that existing nodes always keep updated. They’re not special during normal operation — just initial gossip contacts.
7. Write path [R1, R5]
End-to-end write pipeline
Client
│ put(key, value, context)
▼
Router / Smart Client
│ hash(key) → preference list [A, B, C]
│ forward to coordinator (preferably a node in the preference list)
▼
Coordinator (Node A)
│ 1. Increment version vector: [(A, n+1)]
│ 2. Write to local storage:
│ WAL append (sequential write, fsync → durable)
│ MemTable update (in-memory, fast)
│ 3. Fan out in parallel to B and C
├──────────────────────┬──────────────────────┐
▼ ▼ ▼
Node A (local ack) Node B Node C (down)
│ │ │
│ ack │ ack │ timeout → hinted handoff to Node D
└──────────────────────┘
W=2 acks received → return success to client
Node D (stand-in)
│ stores hint: { target: C, key, value, version }
│ TTL-bounded; replays to C on recovery
▼
Node C recovers → hint replayed → RF restored to 3
Step by step, with N=3, W=2:
- Client sends
put(key, value, context)— thecontextis the version it last read (more in §9). - The routing layer (a smart client library that gossips ring state, or a thin proxy) forwards to a coordinator — ideally a node in the key’s preference list, saving a hop.
- Coordinator increments its entry in the key’s version vector and writes locally: append to the write-ahead log, apply to the memtable. (Each node’s storage engine is an LSM tree — detailed in §11.)
- Coordinator sends the write to the other replicas in the preference list, in parallel.
- As soon as W=2 acknowledgments arrive (its own counts), the coordinator returns success — total latency is one parallel fan-out, no ordering round.
- The third replica’s ack arrives late or never. Late: fine, it’s already durable on two nodes. Never — say node C is down:
Sloppy quorum and hinted handoff. Rather than fail the write (strict quorum) or silently under-replicate, the coordinator writes C’s copy to the next healthy node on the ring — node D — tagged with a hint: “this belongs to C; deliver it when C returns.” D stores it in a separate hint buffer, not its main keyspace, and replays it to C on recovery. The write met W=2 with a stand-in. This is the mechanism that makes “always-writable” literal: the cluster accepts writes as long as any W nodes are up, not any W specific nodes.
Caution
Common mistake: Counting hinted writes as durable replicas forever. A hint is an IOU, not a replica — if node D dies before replaying C’s hint, that copy is gone, and the key silently runs at replication factor 2. Hints are a patch for transient outages (minutes to hours); the durable repair mechanism is anti-entropy (§9). Candidates who present hinted handoff as the whole repair story miss why Merkle trees exist.
8. Read path [R2]
flowchart TD
Client[Client] --> Router[Routing layer]
Router --> Coord[Coordinator · node A]
Coord --> StoreA[(Storage · A)]
Coord -->|"read (parallel)"| NodeB[Node B]
Coord -->|"read (parallel)"| NodeC[Node C]
Coord -->|read repair| NodeC
classDef store fill:#eef2f1,stroke:#2c5f5d,stroke-width:2px,color:#1f2937;
class StoreA store
Step by step, with N=3, R=2:
- Coordinator sends the read to all N replicas (or R plus speculative extras), waits for R=2 responses.
- Each response carries
(value, version_vector). The coordinator compares vectors:- One version descends from the other (strictly newer) → return the newer one.
- Versions are concurrent (neither descends from the other) → return both as siblings, with a merged context. The client resolves (§9).
- Read repair: if any responding replica returned a stale version, the coordinator pushes the winning version back to it asynchronously. Hot keys — the ones read constantly — are thus also the ones repaired fastest, which is exactly the staleness-exposure profile you want.
- Return
(value(s), context)to the client. The context is the opaque version token the client must send back on its nextputto that key.
Quorum arithmetic. The consistency knob is the pair (R, W) against N:
| N | W | R | R+W > N? | Behavior |
|---|---|---|---|---|
| 3 | 2 | 2 | Yes | Read overlaps write — at least one responder has the latest acknowledged write |
| 3 | 1 | 1 | No | Fastest, always-available — reads may miss recent writes entirely |
| 3 | 3 | 1 | Yes | Slow writes, fast consistent reads — one dead replica blocks all writes |
| 3 | 1 | 3 | Yes | Fast writes, slow reads — read-side pays the coordination cost |
R+W > N guarantees the read set and write set intersect — some node in your R
responses acknowledged the last write. It does not make the system linearizable
(sloppy quorums and concurrent writes still produce anomalies); it makes stale reads
the exception instead of the rule. I’d commit to W=2, R=2 as the default and say the
API exposes per-request overrides.
9. Conflict detection and resolution [R1, R2, R3]
Important
Key takeaway: In a leaderless store, conflicts are not a bug — they’re the purchase price of the always-writable contract. The design question is whether you detect them honestly or erase them silently. The full chain: concurrent writes → why physical clocks fail → vector clocks for detection → sibling resolution → vector growth and GC.
Step 1: How concurrent writes arise
A network partition splits the cluster. A shopping-cart key has version [(A,3)] on both sides:
Left side (reachable: A, B) Right side (reachable: C)
Client 1 adds item X Client 2 adds item Y
Coordinator A: version → [(A,4)] Coordinator C: version → [(A,3),(C,1)]
Both writes succeed — neither side was down. The partition heals. Now two replicas hold divergent versions of the same key. This is a conflict, and it must be resolved before the next read returns.
Step 2: Why physical clocks (LWW) fail
The naive fix: compare wall-clock timestamps, keep the latest. Last-write-wins (LWW).
The problem: [(A,4)] was written at 14:00:00.003 on node A; [(A,3),(C,1)] was written at 14:00:00.007 on node C. LWW picks C’s version. Client 1’s item X is silently erased. The system logged no error. The customer sees their cart inexplicably missing an item.
LWW fails because:
- Clock skew is real — NTP synchronizes to ~1–100ms, not microseconds. “Later timestamp” can mean “node C’s clock ran fast.”
- Causality is invisible — LWW can’t distinguish “this write logically supersedes the other” from “these writes are concurrent and both should survive.”
LWW is acceptable only for genuinely overwrite-style data (a session token, a device heartbeat) where the latest physical value is exactly what you want and losing a concurrent write is not data loss. For anything that should be merged, LWW is wrong.
Step 3: Vector clocks for conflict detection
A vector clock is a map from node ID to a logical counter, stored with every value. It encodes causal history, not time.
Rules:
- On write at coordinator X: increment
vc[X]. The new version causally supersedes any version it descends from. - Version
V1descends fromV2iffV1[n] >= V2[n]for all nodes n (and strictly greater for at least one). V1 is newer — safe to discard V2. - Versions are concurrent iff neither descends from the other — some node entry is higher in V1, another is higher in V2. Neither is “newer.” Both must survive as siblings.
Worked example:
Initial state: [(A,3)] (same on all replicas)
After partition:
Left: Client 1 writes via A → [(A,4)]
Right: Client 2 writes via C → [(A,3),(C,1)]
Compare [(A,4)] vs [(A,3),(C,1)]:
A entry: 4 > 3 → left is ahead on A
C entry: 0 < 1 → right is ahead on C
Neither descends from the other → CONCURRENT → siblings
The coordinator returns both siblings to the client with a merged context token.
Step 4: Conflict resolution (client-side reconciliation)
The client receives siblings and must merge them:
Sibling 1: { items: [X, Y] } ← [(A,4)]
Sibling 2: { items: [Y, Z] } ← [(A,3),(C,1)]
Merged: { items: [X, Y, Z] }
The client writes back the merged value with the combined context [(A,5),(C,1)]. This version descends from both siblings — the conflict is closed, and subsequent reads return a single version.
Who resolves: this is a product decision. The API exposes siblings; the application defines the merge function. A shopping cart uses set-union; a counter uses sum; a “last editor wins” document might pick one. The store itself cannot resolve without knowing the semantics — which is why systems like Riak offer pluggable conflict resolvers while Cassandra defaults to LWW (trading correctness for simplicity).
Step 5: Vector clock growth and GC
Vector clocks grow unboundedly — one entry per coordinator that ever handled the key. A popular key written by 50 different coordinators has a 50-entry vector. Two mitigations:
- Truncation (Dynamo approach): cap vectors at ~10 entries, dropping the oldest. The dropped entry can no longer distinguish concurrent vs. causal — it may produce a false sibling (extra reconciliation work). Acceptable: it creates correctness noise, not data loss.
- Server-managed clocks (Riak’s dvv / dotted version vectors): track causality at the server side, not the client side, eliminating spurious growth from client retries. More complex but cleaner for high-churn keys.
Warning
Production reality: Knowing that vectors must be managed, not just used, is a Staff-level detail. Interviewers probe it directly: “what happens to the vector clock for a key that gets written by every node?” The answer is truncation with accepted false-concurrency risk, not unbounded growth.
Deletes are writes (R3)
A delete can’t just remove the record — a replica that missed the delete would “resurrect” the key during repair, because to repair, a missing key and a never-written key look identical. So deletes write a tombstone: a versioned “this key is deleted” marker that propagates like any write and wins version comparisons like any write. Tombstones are garbage-collected only after the repair cycle guarantees every replica has seen them — collect too early and deleted data comes back from the dead.
Fault tolerance: tiered repair
Two failure tiers, two mechanisms:
Transient failure (minutes to hours) → Hinted Handoff:
- Covered in §7. Hints are IOUs, not replicas — they expire if the target node stays down too long.
Permanent failure → Merkle anti-entropy:
Each node maintains a Merkle tree per key range — a hash tree where each leaf summarizes a slice of keys. Two replicas compare trees root-first:
Compare root hashes → match? Done (range in sync, 1 message)
→ mismatch? Recurse into left/right children
→ identify differing leaf ranges
→ stream only those keys
Comparing a multi-gigabyte range costs kilobytes of hash traffic when in sync — which is the common case, which is why the cluster can afford to run repair continuously. When out of sync, only the divergent slices stream — not the full range.
Read repair is an opportunistic optimization layered on top: when a read finds a stale replica among its R responses, the coordinator pushes the newer version back asynchronously. Hot keys — the ones read constantly — are also the ones repaired fastest, which is the right bias.
10. API design [R1, R2, R3]
Three operations, plus the context discipline:
GET /v1/keys/{key}?consistency=quorum|one [R2]
returns: { values: [...], context } # >1 value = unresolved siblings
PUT /v1/keys/{key} [R1]
body: { value, context } # context from the preceding GET
returns: { context }
DELETE /v1/keys/{key} [R3]
body: { context }
returns: { ok }
Two non-obvious decisions:
The context round-trip. context is the version vector, opaque to clients. A
client that read version X and writes with context X tells the store “this write
supersedes X” — concurrent writes are detected instead of erased. A PUT with no
context is a blind write that always creates siblings under concurrency.
Per-request consistency. consistency=one on a read is a deliberate “fast and
possibly stale” — right for a presence indicator, wrong for a checkout. Putting the
dial in the API admits what’s true: consistency requirements belong to the data,
not the cluster.
Caution
Common mistake:
Designing the API as bare get(key) → value / put(key, value) with no version context. It looks cleaner — and it forces the server into last-write-wins, because clients can no longer express “I’m updating this version.” The API shape is what makes honest conflict handling possible; simplify it away and you’ve silently committed to data loss under concurrency.
The API hides everything else — ring topology, replica placement, repair. A client never knows the cluster grew from 150 to 200 nodes. That abstraction boundary is what lets the system evolve to Level 5 (multi-region) without breaking a caller.
11. Data model and storage
Requirements → components
| Requirement | Component | Role |
|---|---|---|
| R1 — put | Coordinator + quorum write | W-of-N acknowledgment, version increment |
| R2 — get | Coordinator + read repair | R-of-N read, sibling detection |
| R3 — delete | Tombstones | Versioned delete markers, GC after repair cycle |
| R4 — elastic scaling | Ring + gossip + vnodes | ~1/n data movement per topology change |
| R5 — durability | WAL + RF=3 + hints + Merkle repair | Acknowledged writes survive node loss |
Access frequency
| Access | Frequency | Drives |
|---|---|---|
| get/put by key | Very high (1.3M ops/sec peak) | Ring lookup must be local (cached state, no metadata hop) |
| Ring state lookup | Every request | Gossip-replicated to all nodes and smart clients |
| Hint writes | Only during replica outages | Separate buffer, bounded with TTL |
| Merkle exchanges | Continuous, low rate | Background priority, throttled below foreground I/O |
Storage technology choices
| Store | Technology | Serves | Why this, not alternatives |
|---|---|---|---|
| Node-local engine | LSM tree (RocksDB-style) | R1, R2, R3 | The write path is append-dominated (every put is a fresh version, deletes are tombstone writes, nothing updates in place) — LSM turns that into sequential I/O. A B-tree’s in-place updates buy faster range scans we declared out of scope |
| Ring + membership state | In-memory, gossip-replicated; seed list persisted | R4 | Kilobytes of data every request consults — must be a local read. Eventual convergence is fine: a briefly stale ring view just means a forwarding hop |
| Hint buffer | Local LSM column family, TTL’d | R5 | Hints are queue-like (write, replay, delete) and must not mix with owned data — a node should never serve a hinted value as its own |
| Merkle trees | Computed per range, cached, rebuilt on compaction | R5 | Derived data — rebuildable, never authoritative, so it needs no replication of its own |
Access pattern matrix
| Req | Access pattern | Structure | Key used | Caller |
|---|---|---|---|---|
| R1 | Append write + version | data (LSM) | key (ring position = hash(key)) | Coordinator |
| R2 | Point read + version compare | data (LSM) | key | Coordinator |
| R3 | Append tombstone | data (LSM) | key | Coordinator |
| R4 | Preference-list lookup | ring state | hash(key) | Router / any node |
| R5 | Buffer missed write | hints | (target_node, key) | Coordinator |
| R5 | Range hash comparison | merkle trees | key range | Repair daemon |
Schemas
Data record [R1, R2, R3]
data (per-node LSM):
key: raw bytes — ring position is hash(key)
value: opaque blob, ≤ 100 KB
version: vector clock [(node_id, counter), ...]
tombstone: boolean
written_at: wall-clock timestamp — for tombstone GC and metrics only,
never for conflict resolution
Warning
Production reality: Version vectors grow — one entry per coordinator that ever handled the key. Long-lived, hot keys accumulate entries until the metadata rivals the value. Dynamo truncates vectors beyond ~10 entries (oldest first), accepting a small chance of false-concurrency (a resolvable sibling) in exchange for bounded metadata. Knowing that vectors must be managed, not just used, is a Staff-level detail interviewers probe.
Ring state [R4] — gossip-replicated to every node:
ring_state:
node_id: uuid
tokens: [position, ...] # ~256 virtual nodes per physical node
status: joining | up | suspect | leaving | down
heartbeat: generation + version counter
Hints [R5] — coordinator-local:
hints:
target_node: uuid # the replica that missed the write
key, value, version: # the write itself
expires_at: timestamp # beyond this, rely on Merkle repair
12. Single-node storage engine
Note
Interview scope: In most interviews, naming the engine and briefly justifying it (one sentence on why LSM over B+Tree given an append-dominated write path) is enough. Go deeper into compaction strategies, bloom filter math, and the Bitcask comparison only if the interviewer asks — this section is reference material for that moment, not required whiteboard content.
Distributed coordination is only half the story. The single-node storage engine is where every get and put ultimately lands — and the wrong engine choice means the system can’t hit the p99 SLA regardless of how well the ring is designed.
LSM-Tree vs B+Tree vs Bitcask
| Engine | Write model | Read model | Good for | Watch out for |
|---|---|---|---|---|
| LSM-Tree (RocksDB, Cassandra) | Append to memtable + WAL; periodic compaction to SSTable | Point read requires bloom filter + multi-level probe | Write-heavy workloads; large datasets; compressible values | Read amplification (check multiple SSTables); compaction I/O spikes competing with foreground |
| B+Tree (InnoDB, PostgreSQL) | In-place page update; WAL for durability | O(log n) tree traversal; excellent for range scans | Read-heavy; range queries; OLTP | Write amplification (random I/O for in-place updates); fragmentation at high write rates |
| Bitcask (Riak default) | Append-only log; in-memory hash index for all keys | Single disk seek (index → offset → value) | Small key sets where all keys fit in RAM; highest point-read throughput | Keys must fit in memory; no range queries; compaction (“merge”) required for space reclaim |
Why LSM wins here
This system’s write path is append-dominated: every put is a fresh versioned record (not an in-place update), deletes are tombstones, and compaction periodically collapses versions. LSM maps perfectly to this access pattern:
Write path (LSM):
1. Append to WAL (sequential write, O(1), crash-safe)
2. Insert into MemTable (in-memory sorted structure, O(log n))
3. When MemTable full → flush to L0 SSTable (sequential write)
4. Background compaction: merge SSTables, drop stale versions and tombstones
Read path (LSM):
1. Check MemTable (in-memory — fast)
2. Check bloom filter per SSTable level (probabilistic, avoids disk read)
3. Binary search in SSTable index → read data block
4. Merge versions across levels if multiple hits
Compaction is the LSM’s maintenance cost: it reclaims space, drops old versions, and improves read performance (fewer levels to check). The two main strategies:
| Strategy | How | Trade-off |
|---|---|---|
| Leveled (RocksDB default) | Each level has a size limit; SSTables within a level are disjoint key ranges | Better read performance (each key in at most one SSTable per level); higher write amplification |
| Size-tiered (Cassandra default) | Merge SSTables of similar size together | Better write throughput; worse read amplification (key may appear in multiple SSTables) |
The practical rule: leveled for read-heavy workloads; size-tiered for write-heavy. This system is close to balanced (§1’s 1M reads / 300K writes) — leveled with a generous compaction throughput budget is the right call.
Bloom filters are the LSM’s read optimization: a probabilistic structure (per SSTable, ~10 bits/key) that answers “does key X definitely NOT exist in this SSTable?” with no false negatives. At 1M reads/sec, bloom filters reduce disk I/O by eliminating the ~80–90% of SSTables that don’t contain the queried key without reading them.
Bitcask: when it’s better
Bitcask wins if the key count fits entirely in RAM (~billions of keys at 64 bytes each = ~64 GB) and the access pattern is nearly all point reads. Its single-seek guarantee (one hash lookup → one disk read) is unbeatable for point-read throughput. Riak uses Bitcask as its default backend for exactly this reason. At 100B keys × 64 bytes of index ≈ 6.4 TB of index — doesn’t fit in RAM — so Bitcask is not viable here.
13. Advanced engineering challenges
Hotspot key mitigation
The ring distributes keys uniformly, not load. A viral key (a global config object, a celebrity’s session) concentrates all its traffic on exactly N=3 nodes. At 1M reads/sec globally, a single hot key can saturate those three nodes while the other 147 are idle.
Mitigation strategies:
-
Key salting (write side): append a random suffix
0..kto the key before writing —config:global#3,config:global#7. Reads scatter-gather from all k suffixes and merge. This breaks the “single logical key” abstraction at the application layer but is the only way to distribute write load across more than N nodes. -
Local cache (read side): the coordinator or client library caches the value for a short TTL (1–5 seconds). At 1M reads/sec on one key, even a 1-second cache reduces backend load by ~99.9%. Acceptable for configuration-style data; wrong for anything requiring strong freshness.
-
Request coalescing: if many threads ask for the same key simultaneously, coalesce them into one backend request and fan the response back to all waiters. This is a coordinator-layer optimization that doesn’t sacrifice consistency.
The operational signal to watch: per-key request rate from hot-key detection middleware. Alert when any single key exceeds 10% of a single node’s capacity — that’s the inflection point where the ring’s uniformity guarantee breaks down.
p99 latency: hedge requests
The Dynamo design collects R responses out of N. With N=3, R=2, and a 10ms p99 target, the expected latency is the faster of two parallel responses. But the tail latency is the faster of two responses where one node is occasionally slow — a GC pause, a compaction burst, a network hiccup.
Hedge requests (also called speculative execution): if the first R-1 responses haven’t returned within a threshold (say, 2ms past the median), issue a request to an additional replica before the R threshold is met. Take whichever R responses arrive first. The slow replica is deprioritized, not waited for.
Timeline without hedging:
t=0ms Send to A, B, C
t=3ms A responds
t=8ms B responds ← R=2 satisfied, return at t=8ms
t=50ms C responds (GC pause)
Timeline with hedging (threshold=5ms):
t=0ms Send to A, B, C
t=3ms A responds
t=5ms No second response yet → hedge to D
t=6ms D responds ← R=2 satisfied, return at t=6ms
t=8ms B responds (ignored)
t=50ms C responds (ignored)
The cost: slightly higher network traffic (one extra request per hedged operation — rare). The payoff: p99 collapses toward p50 because the slow tail is avoided, not waited for. Jeff Dean’s research at Google found hedging reduced p99 latency by 50%+ with <5% extra requests in practice.
Threshold selection: set the hedge threshold at the p50 latency of that request type. Requests that haven’t responded at the median are likely stragglers; issue the hedge. Don’t set it too low (unnecessary extra requests) or too high (no benefit for slow tails).
14. Failure modes
I’d walk through these proactively, grouped by what they threaten. The pattern: name what fails, name what degrades, name what doesn’t.
Availability failures
A replica dies during writes
Fails: one member of the preference list. Degrades: nothing visible — sloppy quorum routes its share to the next ring node as hints; latency unchanged. Doesn’t degrade: write acceptance. The honest caveat I’d name: durability quietly runs at RF=2-plus-IOU until the node returns or repair re-replicates — this is the contract’s fine print.
Coordinator dies mid-request
Fails: the in-flight request. Degrades: the client retries against any other node — every node can coordinate, so there’s no failover wait. Doesn’t degrade: correctness. A retried put with the same context produces a version that supersedes or siblings the first attempt; it cannot half-apply.
Consistency failures
Concurrent writes during a partition
Fails: single-copy illusion — both sides accepted writes to the same key. Degrades: the next read does extra work (returns siblings; client merges, as in §9’s worked example). Doesn’t degrade: data survival — both writes exist. The contrast to name: under LWW this same scenario is silent data loss, and nobody is ever told.
Clock skew
Fails: any meaning “latest timestamp” had. With vector clocks, foreground reads and writes don’t care — causality is counters, not clocks. What’s still exposed: tombstone GC windows and TTLs, which do use wall clocks. Mitigation: NTP discipline and generous GC margins, not clock trust.
Durability and topology failures
A node is lost permanently
Fails: one replica of every range the node’s 256 vnodes owned. Degrades: those ranges run at RF=2 while Merkle repair re-streams them — from dozens of source nodes in parallel, the payoff of virtual nodes. Doesn’t degrade: reads and writes, since quorums still assemble. The operational risk to name: repair I/O competes with foreground traffic; unthrottled, the cure causes the latency outage the failure didn’t.
Gossip partition splits the ring view
Fails: membership agreement — each side suspects the other is down. Degrades: each side keeps serving with sloppy quorums and accumulates hints; conflicting writes pile up for later resolution. Doesn’t degrade: write availability — this is the AP choice, visible in its rawest form. What’s lost during the partition: any pretense of reading globally-latest data. After healing: hint replay + read repair + anti-entropy converge the sides; siblings surface to clients.
Hot key overwhelms its three replicas
Fails: latency on one preference list (a celebrity-config key, a viral object). The ring spreads keys evenly, not load — N=3 specific nodes serve every request for this key. Degrades: with mitigation, freshness — a short-TTL client-side cache or request coalescing absorbs the read storm at the cost of seconds-stale reads. Doesn’t degrade (must not): neighbors — without back-pressure, the three replicas’ overload spills into every other range they host, turning one hot key into a cluster incident.
15. What I’d skip, and say I’m skipping
Time check — five minutes left. Things I’d explicitly defer:
- Storage-engine internals. Compaction strategies, bloom filters, block caches — a full question of its own. I’ve justified why LSM; tuning it is RocksDB’s job.
- Range queries and secondary indexes. They’d force order-preserving partitioning and reopen the hot-range problem — a different design with different tradeoffs.
- Multi-region active-active. The Level 5 evolution: per-region quorums, async cross-region replication, and CRDT value types to make merges automatic. I’d name CRDTs as the direction and stop.
- Transactions. Cross-key atomicity needs consensus or 2PC — exactly the coordination this architecture exists to avoid. If the interviewer wants it, that’s a signal to switch architectures, not bolt it on.
- Security and multi-tenancy. Real products need auth, encryption, per-tenant quotas; none of it shapes this architecture.
Saying “I’d skip this, and here’s why” is a strong signal. It shows you know the full surface and are making deliberate scoping choices.
Interview flow summary
Architecture at a glance
WRITE PATH READ PATH
put(key, value, context) get(key)
↓ ↓
Coordinator (any node) Coordinator (any node)
↓ ↓
hash(key) → preference list (N=3) hash(key) → preference list (N=3)
↓ ↓
Parallel send to replicas Parallel read from replicas
↓ ↓
W=2 acks → success R=2 responses → compare versions
↓ ↓
(replica down → hint to next node) (stale replica → read repair)
↓
(concurrent → return siblings)
The walkthrough order for your whiteboard
1. Requirements & contract — surface "always-writable" — it decides everything
2. Capacity estimate — 150 nodes; failure is steady state, data movement is the cost
3. Architecture evolution — Levels 1–5; commit to Level 4 (leaderless ring)
4. Core choice — Consensus vs Leaderless contrast; justify leaderless by contract
5. Consistent hashing — ring, vnodes, scale-out/in mechanics, gossip propagation
6. Write path — WAL→MemTable→quorum W, sloppy quorum, hinted handoff
7. Read path — quorum R, R+W>N arithmetic, read repair
8. Conflict resolution — concurrent writes → LWW failure → vector clocks → siblings → GC
9. API — context round-trip, per-request consistency dial
10. Data model — LSM rationale, ring/hints/merkle schemas
11. Storage engine — LSM vs B+Tree vs Bitcask; compaction; bloom filters
12. Advanced — hot key mitigation, hedge requests
13. Failure modes — availability / consistency / durability groups
If the interviewer pushes deeper
| Depth level | They’re probing for | Where to go |
|---|---|---|
| Level 1 | Partitioning basics | Ring mechanics, mod-n strawman, virtual nodes (§6) |
| Level 2 | Scaling mechanics | 1/k data movement, vnode parallelism, gossip propagation (§6) |
| Level 3 | Quorum arithmetic | R+W>N table, what overlap does and doesn’t guarantee (§8) |
| Level 4 | Conflict resolution | LWW failure → vector clock chain → sibling reconciliation → GC (§9) |
| Level 5 | Consensus contrast | Raft/Dynamo comparison, partition behavior, conflict semantics (§5) |
| Level 6 | Storage engine | LSM compaction strategies, bloom filters, Bitcask tradeoff (§12) |
| Level 7 | Production hardening | Hotspot mitigation, hedge requests, repair throttling (§13) |
| Level 8 | Beyond one region | Per-region quorums, CRDTs — name it, don’t build it (§15) |
16. Wrap-up
One crisp sentence before the interviewer’s next question:
This design buys always-writable availability by giving up the single-copy illusion — consistent hashing bounds what moves when topology changes, quorum overlap makes staleness the exception, and version vectors ensure that when concurrency does happen, the system surfaces both versions honestly instead of silently discarding one.
What separates levels on this question
- SDE II draws the hash ring, names N/R/W replication, and sketches get/put through a coordinator. Usually treats failure as an afterthought (“we’d detect it and rebalance”).
- SDE III runs the quorum arithmetic and says what R+W>N does not guarantee, explains hinted handoff and why hints aren’t replicas, walks a concrete conflict through vector clocks, chooses LWW vs siblings per value type, and names 4+ failure modes with what degrades in each.
- Staff/Principal frames the availability/consistency contract as a product decision and pins the interviewer down on it before designing; treats repair traffic and rebalance throttling as first-class capacity concerns (the fleet is storage-bound); knows where the metadata bites (vector-clock truncation, tombstone GC resurrection windows); and names the boundary where this architecture stops being the answer — transactions or strong ordering means consensus, not a bigger ring.
The difference isn’t knowledge. It’s honesty about guarantees — weaker candidates claim the quorum makes things consistent; stronger ones can say precisely which anomalies remain and why the contract tolerates them.
Further reading
- Dynamo: Amazon’s Highly Available Key-value Store — the paper this entire design space descends from. Consistent hashing, sloppy quorums, vector clocks, Merkle repair — all of it is here, with production war stories.
- Designing Data-Intensive Applications — Kleppmann. Chapter 5 (replication, including the leaderless section) and Chapter 6 (partitioning) are the best textbook treatment of every mechanism in this walkthrough.
- System Design Interview Vol. 1 — Alex Xu. Chapter 6 is the canonical interview-prep version of this question.
- Werner Vogels: Eventually Consistent — the consistency-model vocabulary (read-your-writes, monotonic reads) for naming exactly what a quorum configuration gives up.
- Cassandra architecture documentation — how a production system industrialized the Dynamo design: gossip, hinted handoff, and tunable consistency as shipped features.
- Riak’s vector clock documentation — the most readable practitioner’s guide to causal context and sibling resolution.