Design a Distributed Key-Value Store

1. Problem & Real-World Context

Design a Dynamo/Cassandra-style distributed key-value store: put(key, value) and get(key), horizontally scalable, surviving node failures without downtime. This is the problem that makes CAP theorem concrete rather than theoretical — every design decision here (replication factor, quorum size, conflict resolution) is a direct, nameable trade-off between consistency and availability, which is exactly why it's asked: it tests whether a candidate can reason about those trade-offs with specifics, not just cite the theorem.

2. Requirements

Functional: put(key, value), get(key) - simple API surface, deliberately no query language, no joins, no transactions across keys (that scope cut is itself part of the design — a KV store optimizes for exactly this narrow access pattern).

Non-functional: high availability over strict consistency (this is an AP system, in CAP terms — see [[CAP Theorem & Consistency Models]]); horizontally scalable (add/remove nodes without downtime or full data reshuffle); tunable consistency per operation (not every read needs the same guarantee).

3. Capacity Estimation

• At, say, 1M writes/sec and 10M reads/sec across a large deployment, with a replication factor of 3, actual per-node write load is (1M × 3) / num_nodes - replication multiplies write cost, which is why replication factor is a deliberate trade-off (durability/availability vs. write cost), not a free durability upgrade.
• Partition count: with consistent hashing (§5) and virtual nodes, the ring is typically divided into far more logical partitions than physical nodes (e.g., 256 vnodes per physical node) — this is what allows adding one node to absorb load from many existing nodes' worth of partitions simultaneously, rather than from just one neighbor.

4. API Design

PUT /v1/kv/{key}   { "value": ..., "consistency": "quorum" | "one" | "all" }
GET /v1/kv/{key}?consistency=quorum

Exposing consistency as a per-request parameter (rather than one fixed system-wide setting) is the concrete form of PACELC's latency-vs-consistency choice (see [[CAP Theorem & Consistency Models]]) - a caller reading a user's shopping cart might accept one (fast, possibly slightly stale); a caller reading an account balance might require quorum or all.

5. Partitioning: Consistent Hashing

Data is distributed across nodes via [[Consistent Hashing]] - full mechanism there, but the short version: hash each node onto a ring at multiple virtual-node points; a key's owner is whichever node is nearest clockwise on the ring from the key's own hash. This is precisely what makes adding/removing a node reassign only ~1/N of keys instead of nearly all of them, and gives natural per-key replica placement (walk clockwise, take the next R distinct physical nodes for a replication factor of R).

6. Replication & the N/W/R Quorum Model

Each key is replicated to N nodes (the replication factor, e.g., N=3). Every write and read specifies how many replicas must participate:

• W = number of replicas that must acknowledge a write before it's considered successful.
• R = number of replicas that must respond to a read before returning a result (the read merges/reconciles their responses, see §7).

The quorum guarantee: if W + R > N, every read is guaranteed to overlap with at least one replica from the most recent successful write — this overlap is what gives strong-ish consistency without needing all N replicas to participate in every operation.

def quorum_satisfied(n: int, w: int, r: int) -> bool:
    return w + r > n

# Common configurations for N=3:
quorum_satisfied(3, 2, 2)   # True  — "quorum" reads/writes, balanced latency/consistency
quorum_satisfied(3, 1, 1)   # False — fast, low latency, but reads may miss the latest write
quorum_satisfied(3, 3, 1)   # True  — write-heavy-safe: slow writes (all must ack), fast reads
quorum_satisfied(3, 1, 3)   # True  — read-heavy-safe: fast writes, reads check every replica

The trade-off table this produces is the actual interview content: W=N, R=1 favors fast reads at the cost of slow, less available writes (every replica must be up to accept a write); W=1, R=N is the mirror image; W=R=majority (e.g., 2-of-3) is the balanced default most systems ship with.

7. Conflict Resolution: Vector Clocks

With W < N, concurrent writes to the same key from different clients (or during a network partition where both sides accept writes) can produce conflicting versions that no amount of quorum math resolves by itself — something has to decide which version "wins," or that both must be kept and reconciled later.

Last-Write-Wins (LWW) — simplest: attach a timestamp to each write, keep the one with the latest timestamp on conflict. Cheap, but silently loses data — if clock skew between nodes is even a few hundred milliseconds (a very real possibility, see the [[WhatsApp Chat]] page's discussion of the same clock-skew problem), an actually-later write can be incorrectly discarded because its originating node's clock ran slightly behind.

Vector clocks — the more correct answer: each value carries a vector of (node_id, counter) pairs, incremented by whichever node handled the write.

def vector_clock_compare(v1: dict[str, int], v2: dict[str, int]) -> str:
    """Returns 'before', 'after', or 'concurrent'."""
    all_nodes = set(v1) | set(v2)
    v1_leq = all(v1.get(n, 0) <= v2.get(n, 0) for n in all_nodes)
    v2_leq = all(v2.get(n, 0) <= v1.get(n, 0) for n in all_nodes)
    if v1_leq and not v2_leq:
        return "before"        # v1 happened-before v2 — v2 is the newer version
    if v2_leq and not v1_leq:
        return "after"         # v1 is newer than v2
    if v1_leq and v2_leq:
        return "same"
    return "concurrent"        # neither dominates — a genuine conflict, can't be resolved by clocks alone

When two versions are genuinely concurrent (neither vector clock dominates the other), the system can't unilaterally decide which is correct — it returns both versions to the client (or application logic) for reconciliation. Dynamo's canonical example: a shopping cart with concurrent adds from two devices — the correct resolution is a merge (union of both carts' items), which only application-level logic can perform correctly; the storage layer's job is surfacing the conflict, not guessing a resolution.

8. High-Level Architecture

flowchart LR
    C[Client] --> Coord[Coordinator Node<br/>any node can coordinate]
    Coord --> N1[Node A - replica]
    Coord --> N2[Node B - replica]
    Coord --> N3[Node C - replica]
    N1 <-.gossip.-> N2
    N2 <-.gossip.-> N3
    N3 <-.gossip.-> N1

Any node can act as the coordinator for a given request — it looks up (via the ring) which nodes own the key, forwards the read/write to those replicas, and reconciles responses according to R/W before replying to the client. There's no single fixed "leader" node for the whole cluster — this decentralization (peer-to-peer, no single point of coordination) is a deliberate availability-maximizing choice, directly opposed to a primary/replica model's stronger consistency guarantees.

Membership/failure detection via gossip: nodes periodically exchange state with a few random peers (not a central registry) — over several rounds, information about a node joining or failing propagates to the whole cluster without requiring a coordinator or single source of truth, matching the same decentralized philosophy as the rest of the design. See [[Service Discovery & Health Checks]] for the general heartbeat pattern this specializes.

Baseline model — heartbeat-counter gossip (the scheme SWIM later improves on; understand this first, because the interviewer often starts here): each node keeps a membership list of (member_id, heartbeat_counter, last_updated_local_time) rows.

1. Each node periodically increments its own heartbeat counter — this counter is a monotonic liveness proof: "I am still alive, and this is how many ticks I've seen."
2. Each node periodically sends its whole membership list to a few random peers, which in turn re-gossip to their random peers — so counters fan out across the cluster in O(log N) rounds, no central collector.
3. On receiving a peer's list, a node merges by taking the max counter per member (a higher counter always means fresher liveness info) and stamps its own local receive-time against any row it just bumped.
4. If some member's counter hasn't advanced for longer than a threshold (measured against local receive-time, e.g. no increase in 10s), that member is declared offline and the news gossips out the same way.

def merge_membership(local: dict[str, Member], incoming: dict[str, Member], now: float) -> None:
    for mid, remote in incoming.items():
        cur = local.get(mid)
        if cur is None or remote.heartbeat > cur.heartbeat:
            local[mid] = Member(mid, remote.heartbeat, last_updated=now)  # fresher counter → adopt + restamp local clock

def detect_failures(local: dict[str, Member], now: float, threshold: float) -> list[str]:
    return [mid for mid, m in local.items() if now - m.last_updated > threshold]  # counter stalled locally → offline

Why compare against local receive-time, not the remote clock: this is the subtle correctness point. The threshold measures "how long since I last saw this counter move," which needs no synchronized clocks across nodes — sidestepping the exact clock-skew trap that sinks Last-Write-Wins (§7).

Where the baseline breaks (and why SWIM exists): the failure signal is purely absence — a stalled counter. A single dropped gossip packet, a GC pause, or a briefly saturated link stops the counter propagating and trips the threshold → false-positive eviction of a perfectly healthy node. There's no second opinion and no "are you sure?" step. Tightening the threshold to detect real deaths faster makes false positives worse; loosening it to avoid false positives makes real-death detection slower. That tension is unwinnable with counters alone — which is precisely what the probe-based scheme below fixes.

How failure is actually detected (SWIM-style protocol) — "gossip" alone only explains propagation; detection itself is a separate, concrete mechanism that replaces the counter-staleness signal with an active probe:

1. Direct probe: every gossip round (e.g., every 1s), a node picks 1 random peer from its membership list and sends it a direct ping, expecting an ack within a timeout.
2. Indirect probe (the key trick): if the direct ping times out, the prober doesn't immediately declare failure — a single dropped packet or GC pause would cause false positives. Instead it asks k other random members (e.g., k=3) to each ping the suspect on its behalf. If any of them get an ack, the target is alive — the direct probe path just had a transient network issue. This is what makes the detector robust to noisy networks instead of a naive "1 timeout = dead."
3. Suspicion state, not instant death: only if all direct and indirect probes fail does the prober mark the peer suspected (not dead yet) and gossips that suspicion to the cluster. A suspected node still has a window (a few gossip rounds) to refute it by responding to any probe — this guards against permanently ejecting a node that was just momentarily slow.
4. Confirmation via gossip dissemination: the suspicion itself piggybacks on the regular gossip exchange (no separate message type) — each infected node relays it to its next random peers, so the suspicion (or a later refutation) spreads in O(log N) rounds. If the suspicion window expires with no refutation, the node is marked dead and that fact propagates the same way.

Why this matters over naive heartbeats: a single central heartbeat monitor is a single point of failure and a bottleneck at scale (O(N) heartbeats to one place); this decentralized scheme is O(1) messages per node per round with detection time scaling ~O(log N) with cluster size, and the indirect-probe step specifically is what prevents transient network blips from being misread as node failure.

Who uses which in production (name-drop this to show it's not textbook-only):

The one-line interview takeaway: Dynamo/Cassandra lineage = heartbeat-counter gossip + Phi Accrual (staleness/statistical signal); HashiCorp/Serf lineage = SWIM (active probe + indirect confirm). Both are decentralized gossip; they differ in how a node is judged dead, which is the distinction the baseline-vs-SWIM comparison above draws out.

9. Read Path & Write Path

Walking one write and one read through the full stack — this is where §5–§7 stop being separate concepts and become one concrete sequence.

Write Path (PUT)

1. Client sends PUT /v1/kv/{key} to any node — that node becomes the coordinator for this request.
2. Coordinator hashes key onto the consistent-hashing ring (§5) and walks clockwise to find the N replica nodes that own it.
3. Coordinator attaches a new vector-clock version to the value — incrementing its own counter, merging in the clock of whatever version it's overwriting if known (§7).
4. Coordinator sends the write to all N replicas in parallel (including itself, if it's one of them).
5. Coordinator waits for W acknowledgments, not all N — the first W replicas to durably persist the write are enough; the remaining N−W continue processing asynchronously.
6. If a target replica is unreachable, hinted handoff (§11) kicks in so W can still be satisfied without blocking on a genuinely down node.
7. Once W acks arrive, the coordinator returns success to the client — replication to the rest continues in the background.

def put(coordinator, key: str, value: bytes) -> bool:
    replicas = ring.get_n_replicas(key, n=N)                              # step 2
    new_clock = vector_clock.increment(coordinator.node_id, existing_clock_for(key))  # step 3
    acks = 0
    for replica in replicas:                                              # step 4
        if replica.send_write(key, value, new_clock, timeout=WRITE_TIMEOUT):
            acks += 1
        else:
            hinted_handoff.stash_for(replica, key, value, new_clock)      # step 6
        if acks >= W:                                                     # step 5
            return True                                                   # step 7 — don't wait for the rest
    return acks >= W

Read Path (GET)

1. Client sends GET /v1/kv/{key} to any node — that node becomes the coordinator.
2. Coordinator hashes key onto the ring, finds the same N replicas that own it.
3. Coordinator sends the read to the replicas in parallel.
4. Coordinator waits for R responses, each carrying its value + vector clock.
5. Coordinator compares the R vector clocks (§7): if one strictly dominates the others, that's the answer. If two are concurrent, both are returned to the caller for application-level reconciliation (Dynamo's sibling-values behavior).
6. If any responding replica's version is stale relative to the winning one, the coordinator opportunistically writes the newer version back to it before replying — read repair (§11), folded into the read path instead of a separate process.

def get(coordinator, key: str) -> list[VersionedValue]:
    replicas = ring.get_n_replicas(key, n=N)                              # step 2
    responses = coordinator.fan_out_read(replicas, key, timeout=READ_TIMEOUT)  # step 3
    responses = wait_for_at_least(responses, r=R)                         # step 4
    winners = vector_clock.resolve(responses)                             # step 5 — may be >1 if concurrent
    for r in responses:
        if is_stale(r.clock, relative_to=winners):
            replica_of(r).repair(key, winners[0])                        # step 6 — read repair
    return winners

Why the write path returns after W acks, not N: waiting for all N means one slow or down replica stalls every write — exactly the availability cost the N/W/R model exists to avoid. The remaining replicas still get the write; the client just doesn't wait on them.

Why the read path can return multiple values: this is the direct consequence of §6's quorum guarantee — it guarantees overlap with a recent write, not that all R responses agree. Reconciling disagreement is a first-class part of the read path, not an edge case bolted on afterward.

10. Per-Node Storage Engine (Commit Log, Memtable, SSTable)

§9 describes fan-out across replicas at the coordinator level; this section is what happens inside one replica node once it actually receives that write or read — the real answer interviewers want when they push past "the coordinator sends it to N nodes" to "then what?" This is modeled on Cassandra's actual storage engine, the same architecture Dynamo-style stores in practice use.

Write path on a single node:

1. Commit log — the write is first appended to an on-disk commit log: a write-ahead, sequential, append-only log (the same durability role as Postgres's WAL — see [[Postgres Internals]]). This makes the write durable and crash-recoverable before anything touches the in-memory structure.
2. Memtable (memory cache) — the write is then applied to an in-memory sorted structure (the memtable). Reads and writes against recent data hit this directly, with zero disk I/O.
3. SSTable flush — once the memtable fills past a threshold, it's flushed to disk as an SSTable (Sorted String Table) — an immutable, sorted list of <key, value> pairs. Immutability is deliberate: once written, an SSTable is never modified in place, only eventually merged during compaction (below).

flowchart LR
    W[Write arrives<br/>at this node] --> CL[Append to<br/>Commit Log]
    CL --> MT[Apply to<br/>Memtable]
    MT -->|memtable full /<br/>threshold hit| Flush[Flush to<br/>SSTable on disk]

Why a commit log and a memtable, not just one: the memtable alone isn't durable — a crash before flush loses everything still in memory. The commit log alone isn't queryable at speed — it's an append-only sequence, not a sorted structure. Together: the commit log guarantees durability (replayed on crash recovery to rebuild the memtable), while the memtable gives fast in-memory reads/writes without touching the commit log during normal operation.

Read path on a single node:

1. Check the memtable first — if the key is there (the most recent writes are), return immediately, no disk I/O. This is the fast path Figure 6-20-style diagrams show: memory cache hit → return to client directly.
2. If not in the memtable, check on-disk SSTables — but the key could be in any of several accumulated SSTables, so a Bloom filter per SSTable is checked first to cheaply rule out SSTables that definitely don't contain the key (the same never-false-negative, sometimes-false-positive mechanism as [[Web Crawler]]'s dedup filter, applied here to skip disk reads instead of skipping already-crawled URLs).
3. For SSTables the Bloom filter didn't rule out, the key is looked up directly (each SSTable is internally sorted, so this is a fast indexed lookup, not a linear scan) — checking the most recently flushed SSTable first, since that's most likely to hold the latest value if the key was updated after an earlier flush.

flowchart TD
    R[Read arrives] --> MT{In memtable?}
    MT -->|yes| Return[Return value<br/>to client]
    MT -->|no| BF{Bloom filter:<br/>maybe present?}
    BF -->|no, for this SSTable| Skip[Skip it —<br/>zero disk I/O]
    BF -->|yes, for this SSTable| Lookup[Indexed lookup<br/>in that SSTable]
    Lookup --> Return

Compaction (the natural follow-up question): over time, the same key can end up in multiple SSTables — written, flushed, updated, flushed again. A background compaction process merges multiple SSTables into one, keeping only the most recent value per key and reclaiming space from overwritten or deleted entries — the same "background reconciliation, not on the request hot path" philosophy as anti-entropy repair (see [[Merkle Trees]]), just operating within one node's own SSTables instead of across replicas.

Name it: commit log + memtable + SSTable + compaction is an LSM-tree (Log-Structured Merge-tree) — naming it explicitly signals you recognize this as the standard storage engine underlying Cassandra, RocksDB, and LevelDB, not an ad hoc design.

11. Handling Failures

• Hinted handoff: if a replica that should receive a write is temporarily down, another node accepts the write on its behalf (with a "hint" that it belongs to the down node) and forwards it once the original node recovers — preserves write availability (W can still be satisfied) even during a transient node outage, at the cost of temporarily having the data live somewhere other than its "true" ring position.
• Read repair: when a read's quorum responses disagree (some replicas have a stale version), the coordinator detects this via version comparison (§7) and pushes the newer version back to the stale replicas as part of serving the read — a cheap, incremental way to fix drift without a separate repair process for every inconsistency.
• Anti-entropy (Merkle trees): a background process comparing replica datasets via Merkle tree hashing to catch and repair divergence that read repair alone misses (data that's never read won't trigger read repair) — see [[Merkle Trees]] for the full mechanism (root-hash comparison, pruning matching subtrees, O(log n) vs O(n)).

12. Deep Dives / Follow-ups

• Sloppy quorum: during a partition, a coordinator might not be able to reach N "natural" replicas at all — sloppy quorum relaxes strict replica-set membership (accept writes from any N reachable nodes, reconciled later via hinted handoff) purely to preserve availability, at a further consistency cost beyond the base quorum model.
• Tunable consistency in practice: real systems (Cassandra) expose this almost exactly as consistency=ONE/QUORUM/ALL per query — validating that the N/W/R model isn't a textbook abstraction but literally how production tunable-consistency stores are configured.
• Range queries: a pure KV store's partitioning-by-hash makes range scans (e.g., "all keys between X and Y") expensive/impossible without a secondary index, since consistent hashing deliberately scatters sequential keys across the ring — worth naming as an explicit limitation, not silently ignoring.

13. Interview Drill Questions

1. Why is W + R > N the condition for quorum consistency, not just W = N and R = 1? — W = N requires every replica to be reachable for every write, tanking availability the moment any single replica is briefly down; W + R > N guarantees read/write overlap (so reads see the latest write) while allowing both W and R to be less than N, preserving availability against individual replica outages — this is the actual mechanism behind "tunable consistency," not just a slogan.
2. Two clients write to the same key concurrently during a partition. How does the system avoid losing one write? — Vector clocks tag each write with per-node version info; if the two writes are genuinely concurrent (neither vector clock dominates), both versions are kept and surfaced to the client/application for reconciliation, rather than the system silently picking one (as naive Last-Write-Wins would, risking data loss under clock skew).
3. A node was down for an hour and just came back online. How does it catch up? — Hinted handoff writes that accumulated on other nodes on its behalf are forwarded to it; anti-entropy (Merkle-tree comparison, §11) catches any remaining divergence not covered by hints, particularly for data that wasn't written during the outage but existed before it and might have drifted.
4. Walk through what happens if the coordinator receives 2 acks quickly, W=2, but the 3rd replica's ack arrives late. — The write already returned success to the client the moment the 2nd ack arrived (§9) — the write path doesn't wait for N, only W. The 3rd replica's write still lands (just later), bringing the key to full replication; if it never arrives at all (replica down), hinted handoff (§11) covers it instead.
5. A read for a key that was just written moments ago misses in the memtable. Is that possible, and what happens? — No — a just-written key is guaranteed to be in the memtable at the node that accepted the write, since the write path always applies to the memtable before any flush could occur (§10). A miss would only happen if the read landed on a different replica than the one(s) whose acks satisfied W, which is exactly the "reads might miss the latest write" risk quorum overlap (§6) is designed to bound — not a storage-engine bug.
6. Why can't this system efficiently support "give me all keys starting with 'user_123_'"? — Consistent hashing partitions by the hash of the key, which deliberately scatters lexicographically-adjacent keys across unrelated ring positions — a range/prefix query would need to touch nearly every partition, defeating the point of partitioning; this is a genuine architectural limitation, not a missing feature that could be bolted on cheaply.
7. How would you decide N, W, and R for a specific use case (e.g., a session token store vs. a financial ledger)? — Session tokens tolerate a stale read (a slightly-out-of-date "is this session valid" check is low-risk) — favor lower R/W for speed. A ledger cannot tolerate losing a write or serving a stale balance — favor W=majority or W=N and R=majority, explicitly accepting the availability/latency cost, and state that trade-off out loud rather than picking a default without justification.

Cross-links

[[Consistent Hashing]] · [[Merkle Trees]] · [[Postgres Internals]] · [[Web Crawler]] · [[CAP Theorem & Consistency Models]] · [[Service Discovery & Health Checks]] · [[Database Replication & Sharding]] · [[SQL vs NoSQL]]