Skip to main content

Walkthrough: Designing a Distributed Key-Value Store

A full candidate's-eye walkthrough of the Dynamo-style key-value store question — consistent hashing, quorum reads and writes, vector clocks, hinted handoff, and anti-entropy repair.

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:

  1. 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.
  2. 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.
  3. 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:

LevelArchitectureTrigger for next level
1. Single nodeOne server: hash map + write-ahead logDataset exceeds one machine; SPOF
2. Primary/replicaOne writer, read replicas, failoverWrite throughput ceiling; failover drops writes
3. Sharded primariesHash/range shards, each with a primaryResharding pain; per-shard failover still blocks writes
4. Leaderless ringConsistent hashing, quorum R/W, any node coordinatesMulti-region latency; conflict semantics outgrow vector clocks
5. Multi-region / CRDTsActive-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)
ConsistencyStrong (CP) — linearizable by defaultTunable (AP) — eventual by default, strong at R=N cost
Leader roleOne elected leader per partition range; all writes go through itNo leader — any node coordinates any request
Write pathClient → leader → Raft majority quorum → commit → apply → ackClient → coordinator → parallel fan-out to W replicas → ack
Write latencyMajority round-trip on every write (1 extra RTT for consensus log)W parallel acks only (no ordering round)
Network partitionMinority partition refuses writes — no leader, no quorumBoth partitions accept writes — sloppy quorum with stand-ins
ConflictsImpossible — consensus orders all writesPossible — must detect (vector clocks) and resolve (client merge)
FailoverRaft election on leader death (~seconds) — writes block during electionNone — coordinators are interchangeable, no election exists
Read repairNot needed — leader is always authoritativeRequired — replicas diverge; reads detect and heal staleness
Best forFinancial 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 failureBlocks during failover (seconds)Minority side refuses writesContinues if any W stand-ins reachable
Read consistencyStrong from primary; stale from replicasStrong (linearizable)Tunable via R+W vs N
Write latency1 node + async replicationMajority round-trip on every writeW parallel acks (no ordering round)
ConflictsNone (single writer)None (consensus orders)Possible — must detect and resolve
Operational complexityFailover tooling, primary placementConsensus tuning, group rebalancingConflict semantics, repair pipelines

Architecture decisions

DecisionChosenRejectedRationale
Coordination modelLeaderless quorumPrimary/partition, Raft groupsAlways-writable contract; no failover window exists to manage
PartitioningConsistent hashing with virtual nodesRange partitioningPoint lookups only (no range scans in scope); uniform key distribution; minimal data movement on topology change
MembershipGossip protocolCentral config service (ZooKeeper)A membership master reintroduces the SPOF the leaderless design just removed
Conflict resolutionVector clocks + client mergeLast-write-wins timestampsLWW silently drops concurrent writes; clock skew makes “last” a lie
Failure patchSloppy quorum + hinted handoffStrict quorum (fail the write)Strict quorum violates the always-writable contract on a single node failure

Core data structures

StructureLives whereRole
Hash ringEvery node + smart clients (via gossip)Maps hash(key) → preference list of N replicas
Version vectorStored with every valueDetects concurrent writes vs causally-ordered ones
Hint bufferCoordinator-localHolds writes owed to temporarily-dead replicas
Merkle treePer node, per key rangeCheap 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

  1. Hash the entire key space onto a ring — say the range 0 to 2^128 − 1, wrapping around. Consistent hashing means both keys and nodes are hashed onto the same ring.
  2. Each node owns one or more tokens (positions on the ring). A key belongs to the first node found walking clockwise from hash(key).
  3. The preference list for a key is the next N distinct physical nodes clockwise. With N=3, key K at position 0x6A… 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:

  1. Every second, each node selects a few random peers and exchanges its full ring state (node IDs, token positions, heartbeat counters, node status).
  2. Information spreads epidemically: after O(log N) rounds (~7 rounds in a 150-node cluster), every node has seen the update.
  3. 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:

  1. Client sends put(key, value, context) — the context is the version it last read (more in §9).
  2. 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.
  3. 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.)
  4. Coordinator sends the write to the other replicas in the preference list, in parallel.
  5. As soon as W=2 acknowledgments arrive (its own counts), the coordinator returns success — total latency is one parallel fan-out, no ordering round.
  6. 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:

  1. Coordinator sends the read to all N replicas (or R plus speculative extras), waits for R=2 responses.
  2. 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).
  3. 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.
  4. Return (value(s), context) to the client. The context is the opaque version token the client must send back on its next put to that key.

Quorum arithmetic. The consistency knob is the pair (R, W) against N:

NWRR+W > N?Behavior
322YesRead overlaps write — at least one responder has the latest acknowledged write
311NoFastest, always-available — reads may miss recent writes entirely
331YesSlow writes, fast consistent reads — one dead replica blocks all writes
313YesFast 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:

  1. Clock skew is real — NTP synchronizes to ~1–100ms, not microseconds. “Later timestamp” can mean “node C’s clock ran fast.”
  2. 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 V1 descends from V2 iff V1[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:

  1. 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.
  2. 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

RequirementComponentRole
R1 — putCoordinator + quorum writeW-of-N acknowledgment, version increment
R2 — getCoordinator + read repairR-of-N read, sibling detection
R3 — deleteTombstonesVersioned delete markers, GC after repair cycle
R4 — elastic scalingRing + gossip + vnodes~1/n data movement per topology change
R5 — durabilityWAL + RF=3 + hints + Merkle repairAcknowledged writes survive node loss

Access frequency

AccessFrequencyDrives
get/put by keyVery high (1.3M ops/sec peak)Ring lookup must be local (cached state, no metadata hop)
Ring state lookupEvery requestGossip-replicated to all nodes and smart clients
Hint writesOnly during replica outagesSeparate buffer, bounded with TTL
Merkle exchangesContinuous, low rateBackground priority, throttled below foreground I/O

Storage technology choices

StoreTechnologyServesWhy this, not alternatives
Node-local engineLSM tree (RocksDB-style)R1, R2, R3The 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 stateIn-memory, gossip-replicated; seed list persistedR4Kilobytes 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 bufferLocal LSM column family, TTL’dR5Hints 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 treesComputed per range, cached, rebuilt on compactionR5Derived data — rebuildable, never authoritative, so it needs no replication of its own

Access pattern matrix

ReqAccess patternStructureKey usedCaller
R1Append write + versiondata (LSM)key (ring position = hash(key))Coordinator
R2Point read + version comparedata (LSM)keyCoordinator
R3Append tombstonedata (LSM)keyCoordinator
R4Preference-list lookupring statehash(key)Router / any node
R5Buffer missed writehints(target_node, key)Coordinator
R5Range hash comparisonmerkle treeskey rangeRepair 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

EngineWrite modelRead modelGood forWatch out for
LSM-Tree (RocksDB, Cassandra)Append to memtable + WAL; periodic compaction to SSTablePoint read requires bloom filter + multi-level probeWrite-heavy workloads; large datasets; compressible valuesRead amplification (check multiple SSTables); compaction I/O spikes competing with foreground
B+Tree (InnoDB, PostgreSQL)In-place page update; WAL for durabilityO(log n) tree traversal; excellent for range scansRead-heavy; range queries; OLTPWrite 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 keysSingle disk seek (index → offset → value)Small key sets where all keys fit in RAM; highest point-read throughputKeys 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:

StrategyHowTrade-off
Leveled (RocksDB default)Each level has a size limit; SSTables within a level are disjoint key rangesBetter read performance (each key in at most one SSTable per level); higher write amplification
Size-tiered (Cassandra default)Merge SSTables of similar size togetherBetter 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:

  1. Key salting (write side): append a random suffix 0..k to 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.

  2. 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.

  3. 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 levelThey’re probing forWhere to go
Level 1Partitioning basicsRing mechanics, mod-n strawman, virtual nodes (§6)
Level 2Scaling mechanics1/k data movement, vnode parallelism, gossip propagation (§6)
Level 3Quorum arithmeticR+W>N table, what overlap does and doesn’t guarantee (§8)
Level 4Conflict resolutionLWW failure → vector clock chain → sibling reconciliation → GC (§9)
Level 5Consensus contrastRaft/Dynamo comparison, partition behavior, conflict semantics (§5)
Level 6Storage engineLSM compaction strategies, bloom filters, Bitcask tradeoff (§12)
Level 7Production hardeningHotspot mitigation, hedge requests, repair throttling (§13)
Level 8Beyond one regionPer-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