Design a Dynamo-style key-value store

TL;DR

• A Dynamo-style store is a leaderless, partitioned, replicated hash table: a key hashes to a ring position, lives on the next N distinct nodes (its preference list), and any of them can coordinate a request. Availability wins — it stays writable through failures, and conflicts are resolved afterwards.
• The cruxes an interviewer probes: (1) consistent hashing with virtual nodes and preference lists, (2) tunable N/W/R quorums, (3) versioning with vector clocks or last-writer-wins plus read repair, (4) failure handling — gossip, sloppy quorums, hinted handoff, Merkle anti-entropy, (5) the node write path from WAL to memtable to SSTable.
• It serves 1B keys at 100k reads/s and 10k writes/s on ~30 commodity nodes: N=3, W=2, R=2, p99 read under 10 ms.

Problem statement and clarifying questions

"Design a key-value store like Dynamo, Cassandra or Riak: put(key, value) and get(key) over opaque keys, horizontally scalable, writable while nodes are down." The answers decide two forks: AP or CP under a partition, and who resolves conflicts.

Question	Assumption taken
Availability or consistency under a partition?	Availability: a cart must accept "add item" during an outage and reconcile later; W and R tune it per request.
Value size and shape?	Opaque blobs, ~1 KB median, 1 MB maximum; no secondary indexes, no range scans.
Scale?	1B keys, 10B reads/day, 1B writes/day (10:1 reads to writes).
Latency target?	p99 read < 10 ms, p99 write < 20 ms in-region.
Durability?	An acknowledged write is on two nodes' write-ahead logs; N=3 copies on distinct, rack-aware nodes.
Who resolves conflicting writes?	The application merges siblings; last-writer-wins is opt-in per keyspace.
Multi-datacenter?	Single region first; multi-region replication is a v2 step.
Membership changes?	Explicit: an operator adds or removes nodes; failures are detected automatically.
Transactions or multi-key operations?	No. Single-key only; a batch get is a convenience fan-out.

Requirements

Functional

• put(key, value, context) writes; get(key) returns every concurrent version plus an opaque context; delete(key) writes a tombstone.
• Any replica of a key accepts reads and writes; the client chooses W and R within N per request.
• Nodes join or leave without downtime; only ~1/N of the keys move.

Non-functional

• Availability: 99.99% for writes, with one node down or one rack partitioned away.
• Latency: p99 read < 10 ms and write < 20 ms in-region.
• Durability: N=3, W=2 by default; an acknowledged write is fsync-ed to two write-ahead logs.
• Scale: 1B keys x 1 KB = 1 TB logical, 3 TB raw; 100k reads/s, 10k writes/s, 3x at peak.
• Consistency: eventual by default; W + R > N returns the latest acknowledged write while all replicas are reachable (sloppy quorums weaken it; deep dive 4).

Out of scope

Multi-key transactions, secondary indexes, range queries, a query language, authentication and encryption, cross-region conflict policies.

Estimation

Using the latency and estimation tables (a day is ~10^5 s, peak is 3x average):

Quantity	Arithmetic	Result
Read QPS	10B reads/day / 10^5 s	100k/s average, 300k/s peak
Write QPS	1B writes/day / 10^5 s	10k/s average, 30k/s peak
Replica operations at peak	30k writes x N=3; 300k reads x R=2	90k replica writes/s, 600k replica reads/s
Dataset	1B keys x 1 KB (key, value, clock)	1 TB logical, x3 = 3 TB raw
Storage growth per year	0.1% of writes create keys: 1M/day x 1 KB x 365	~0.4 TB/year logical, ~1.1 TB raw — a third of the dataset per year
Bandwidth	300k reads/s x 1 KB; 30k writes/s x 1 KB x 3 internal	300 MB/s out, 90 MB/s internal: 2.4 Gbps, inside a 10 Gbps NIC
Hot set (80/20 rule)	20% of 1B keys x 1 KB	200 GB, ~7 GB per node of 30: row and page cache, no cache tier
Nodes	90k replica writes/s at ~5k-10k per node = 9-18; x1.5-2 headroom	~30 nodes, ~100 GB each (write throughput binds, not disk)

Say it out loud: the dataset is small but the operation rate is high, so the cluster is sized by replica operations per second, not by disk — each write is N replica writes, each read R replica reads.

API design

Endpoint	Request	Response	Notes
GET /v1/keys/{key}?r=2	—	200 {versions: [{value, clock}], context}	Every sibling; context is an opaque base64 vector clock. 404 when only tombstones remain.
PUT /v1/keys/{key}?w=2	{value} + header X-Context	200 {clock, acked_by}	With a context the write supersedes what that read saw; without one it is blind and may create a sibling. A retry with the same context supersedes its first attempt.
DELETE /v1/keys/{key}	header X-Context	204	Tombstone: replicates like any version, collected after a grace period.
POST /v1/keys:batch-get	{keys: [..]} (up to 100)	200 {items}	Fan-out per key, no atomicity. No pagination: there are no scans.
GET /v1/admin/ring	—	200 {version, nodes: [{id, tokens, status}]}	Clients cache the ring and route straight to a replica, saving a hop (~500 µs).
POST /v1/admin/nodes	{node_id, address}	202	Membership change, spread by gossip; the node streams its ranges in before serving.

r and w are bounded by N; the coordinator rejects combinations it cannot satisfy with healthy replicas (503) rather than silently lowering them.

Data model

The store is its own database; this is the metadata each node keeps about keys, replicas and repairs.

erDiagram
    NODE ||--o{ VNODE : places
    VNODE ||--o{ OBJECT : "owns the range ending at its token"
    OBJECT ||--|{ VERSION : "has one or more siblings"
    NODE ||--o{ HINT : "holds for a node that was down"
    NODE ||--o{ SSTABLE : stores
    NODE {
        string node_id PK
        string address
        string status "up, down or leaving"
        timestamp last_heartbeat
    }
    VNODE {
        bigint token PK
        string node_id FK
    }
    OBJECT {
        string key PK
        bigint token "hash(key) on the ring"
    }
    VERSION {
        string key FK
        string clock PK "vector clock, serialised"
        blob value
        bool tombstone
        timestamp written_at
    }
    HINT {
        string holder_id PK
        string home_id PK
        string key PK
        string clock
    }
    SSTABLE {
        int table_id PK
        string node_id FK
        string min_key
        string max_key
        string bloom_filter
    }

Store choices, one sentence each:

• Partition key is hash(key): the token decides the preference list and nothing else is queried, so no sort key and no secondary index.
• Per-node engine is an LSM tree (RocksDB style): appends on write; reads check the memtable, then Bloom filters, then one block per SSTable (deep dive 5).
• Siblings live under one key as (clock, value) pairs; a read returns them all and the next write collapses them.
• Hints sit in a separate column family keyed by (home node, key), so a recovered node's backlog streams without scanning data files.
• Membership and tokens are a small gossip-replicated table; every node holds the full ring, which is why any node can coordinate.

High-level design

v1: symmetric nodes, a ring-aware client library, and background repair machinery.

flowchart LR
    subgraph clients["Clients"]
        c_app["Application + client library<br/>caches the ring, routes to a replica"]
    end
    subgraph edge["Edge"]
        e_lb["Load balancer<br/>for ring-unaware clients"]
    end
    subgraph services["Services (every storage node runs all of them)"]
        s_coord["Request coordinator<br/>N/W/R state machine"]
        s_ring["Ring + membership<br/>gossip, failure detection"]
    end
    subgraph async["Async"]
        a_hints[["Hinted-handoff queue"]]
        a_repair["Anti-entropy: Merkle exchange<br/>+ compaction"]
    end
    subgraph data["Data"]
        d_n1[("Node 1: WAL, memtable, SSTables")]
        d_n2[("Node 2")]
        d_n3[("Node 3")]
    end
    c_app --> s_coord
    c_app --> e_lb --> s_coord
    s_coord --> s_ring
    s_coord --> d_n1
    s_coord --> d_n2
    s_coord --> d_n3
    s_coord --> a_hints --> d_n1
    a_repair --> d_n2
    a_repair --> d_n3

Write path: the coordinator stamps a clock, writes locally, and acknowledges at W.

sequenceDiagram
    autonumber
    actor CL as Client
    participant CO as Coordinator C
    participant RA as Replica A
    participant RD as Replica D
    participant RE as Stand-in E
    CL->>CO: put(key, value, context)
    CO->>CO: clock = context with C incremented
    CO->>CO: append WAL, update memtable (ack 1 of W=2)
    par send to the other N-1 replicas
        CO->>RA: store(key, version)
        CO->>RD: store(key, version)
    end
    RA-->>CO: ack (W=2 reached)
    CO-->>CL: 200 {clock}
    alt D is up
        RD-->>CO: ack (late, counted for durability only)
    else D is down
        CO->>RE: store(key, version) with a hint for D
        RE-->>CO: ack
        Note over RE,RD: E hands the write to D once gossip marks D up
    end

Read path: merge R answers by clock, return siblings and a context, repair the stale replica.

sequenceDiagram
    autonumber
    actor CL as Client
    participant CO as Coordinator C
    participant RA as Replica A
    participant RD as Replica D
    CL->>CO: get(key), R=2
    CO->>CO: local read: memtable, Bloom filters, SSTables
    par ask the other replicas
        CO->>RA: read(key)
        CO->>RD: read(key)
    end
    RA-->>CO: versions [{A:1, C:1}]
    CO->>CO: R=2 answers: drop dominated versions, merge clocks
    CO-->>CL: 200 {versions, context}
    RD-->>CO: versions [{C:1}] (late, stale)
    CO-)RD: read repair: store the winning versions

Walk-through: the coordinator is whichever replica the client library picked, so there is no leader to elect. A write costs one local WAL append plus one parallel round trip, answered at the W-th acknowledgement; a read adds a clock comparison, not a value merge. Hint handoff, Merkle comparisons and compaction run off the request path.

Deep dive: consistent hashing and preference lists

The probing question is "which nodes hold a key, and what moves when you add a node?"

Scheme	Lookup	Keys moved on +1 node	Balance	Problem
hash(key) mod N	O(1)	~N/(N+1), almost everything	Good	Adding a node reshuffles the cluster
Range partitioning	O(log ranges)	Only the split range	Needs rebalancing	Hot ranges; sequential keys land on one node
Consistent-hash ring, one point per node	O(log N)	~1/N	Poor: some own 2x the mean arc	A removed node dumps its arc on one successor
Ring with virtual nodes	O(log V)	~1/N, spread over all nodes	Good with ~100 points per node	The chosen design

Every node takes ~100 pseudo-random tokens on a 2^32 ring and a key belongs to the first token clockwise from hash(key), which evens out load and spreads a departing node's arcs over many successors (partitioning page).

Specific to this store is the preference list: the walk continued to the first N distinct physical nodes. Skipping further tokens of a node already chosen is the detail candidates forget — without it two of three replicas sit on one machine. Production rings skip same-rack nodes too and run to N + 2 for healthy fallbacks. The hash must be stable across processes and languages, since every client computes the ring; a salted hash() is the wrong tool.

ring = HashRing(["A", "B", "C", "D", "E"], vnodes=64)
ring.get_node("cart:42")  # the owner: first token clockwise from hash(key)
ring.preference_list("cart:42", replicas=3)  # the N distinct nodes holding the replicas

Say the numbers: a sixth node on a five-node ring moves ~1/6 of the keys, all onto the new node; mod N would move ~5/6.

Deep dive: tunable N/W/R quorums

The probing question is "what do W and R buy, and what does W + R > N guarantee?"

Setting	Write latency	Read latency	What you get	Typical use
N=3, W=1, R=1	Fastest	Fastest	Stale reads likely; one replica holds the only copy	Caches, counters you can lose
N=3, W=2, R=2	One round trip, second-fastest ack	Same	Quorums overlap: the latest acknowledged write is in the answer set	The default
N=3, W=3, R=1	Slowest, fails if any replica is down	Fastest	Read-heavy keys; writes stall on one slow replica	Configuration data
N=3, W=1, R=3	Fastest	Slowest, fails if any replica is down	Write-heavy keys	Event ingestion

The overlap argument is the whole point: with W + R > N the write and read quorums share a node, so the newest version is among the R answers. With W + R <= N a reader can contact only replicas that missed the write — the stale-read demo on the replication page; KVCluster.overlapping runs that check. The coordinator sends to all N and waits for W (or R), so tail latency becomes a race the slowest replica loses, at the cost of N replica operations per request.

Two nuances worth volunteering:

• Durability is W, not N. When the client hears "ok" two WALs hold the write; the third copy arrives via a late acknowledgement or a hint.
• The budget is one round trip. A write is an fsync-ed WAL append plus one same-datacenter hop (~500 µs) for the second acknowledgement, a read ~16 µs of SSD per SSTable consulted — both far inside the 10/20 ms p99.

Deep dive: versioning with vector clocks and read repair

The probing question is "two clients update the same cart while a node is down; what does the next reader see?"

Strategy	Concurrent writes	Clock skew	Metadata per version	Burden
Last-writer-wins (wall-clock timestamp)	One silently lost	A fast clock wins every conflict for hours	8 B	Hidden data loss
Vector clocks (per-coordinator counters)	Both kept as siblings	Immune	A few (node, counter) pairs	Application must merge siblings
CRDTs (merge is part of the type)	Merged deterministically	Immune	Type-dependent	Only for data with a natural merge

Choose vector clocks where a lost update matters (carts, profiles) and last-writer-wins per keyspace where it does not (metrics, heartbeats). Cassandra chose LWW, Dynamo and Riak chose clocks; name the trade rather than picking a side by reflex.

Every version is stamped {coordinator: counter, ...}, incremented on the context the client sent. If clock A has every counter of B at least as high, A descends from B and B is dropped; otherwise the writes were concurrent and both survive as siblings. A read returns every sibling plus their merged clock, and the next write carrying that context collapses them: {A:1, C:1} and {C:1, D:1} fold into one version under {A:1, C:1, D:1}.

code/hld/kv_cluster.py — clocks, versions and reconciliation

@dataclass(frozen=True, slots=True)
class VectorClock:
    """One counter per coordinator node, stored as sorted ``(node, counter)`` pairs.

    ``a.descends_from(b)`` means the write stamped ``a`` saw everything the write stamped
    ``b`` saw. When neither clock descends from the other the writes were concurrent, and
    the store keeps both versions as siblings instead of guessing a winner.
    """

    entries: tuple[tuple[str, int], ...] = ()

    def counter(self, node: str) -> int:
        return dict(self.entries).get(node, 0)

    def increment(self, node: str, seen: int = 0) -> VectorClock:
        """Bump ``node``'s counter past both this clock and ``seen``, the highest counter the
        coordinator already stores for the key, so a new version is never dominated by an
        older one the coordinator knows about."""
        counters = dict(self.entries)
        counters[node] = max(counters.get(node, 0), seen) + 1
        return VectorClock(tuple(sorted(counters.items())))

    def merge(self, other: VectorClock) -> VectorClock:
        """The smallest clock that descends from both: the context a reader hands back."""
        counters = dict(self.entries)
        for node, count in other.entries:
            counters[node] = max(counters.get(node, 0), count)
        return VectorClock(tuple(sorted(counters.items())))

    def descends_from(self, other: VectorClock) -> bool:
        return all(self.counter(node) >= count for node, count in other.entries)

    def dominates(self, other: VectorClock) -> bool:
        return self != other and self.descends_from(other)

    def concurrent_with(self, other: VectorClock) -> bool:
        return not self.descends_from(other) and not other.descends_from(self)

    def __str__(self) -> str:
        return "{" + ", ".join(f"{node}:{count}" for node, count in self.entries) + "}"


@dataclass(frozen=True, slots=True)
class Version:
    """A value and the clock of the write that produced it; ``None`` is a tombstone."""

    value: str | None
    clock: VectorClock


def reconcile(versions: Iterable[Version]) -> list[Version]:
    """Drop every version whose clock is dominated by another; what is left is concurrent.

    Two versions with equal clocks but different values (two writes from the same stale
    context through the same coordinator) are both kept rather than silently merged; dotted
    version vectors remove that corner case in production systems.
    """
    unique = sorted(set(versions), key=lambda v: (v.clock.entries, v.value or ""))
    return [v for v in unique if not any(o.clock.dominates(v.clock) for o in unique)]

Read repair is the cheap half of anti-entropy: when one of the R replicas returns a dominated version or nothing, the coordinator writes the winners back after answering the client, so hot keys converge on their own and only cold keys need deep dive 4's Merkle sweep. Clocks are truncated too — Dynamo keeps the ten newest entries, bounding metadata at the price of false siblings.

Deep dive: failure handling with gossip, sloppy quorums, hinted handoff and anti-entropy

The probing question is "a node dies mid-afternoon; walk me through the next hour."

Mechanism	Detects or fixes	Time scale	Cost
Gossip + failure detector	Who is up, who owns which tokens	Seconds	O(log N) rounds, a few messages per node per second
Sloppy quorum	Keeps writes flowing while a home replica is down	Immediate	The write lands on a stand-in outside the list
Hinted handoff	Returns stand-in writes to the home replica	Minutes, on its return	Hint storage on the stand-in, bounded by a TTL
Read repair	Stale replicas of hot keys	On the next read	One extra write per stale answer
Merkle anti-entropy	Cold keys whose hints were lost or expired	Hours, scheduled	Hashes compared in proportion to differences

Two details the table cannot show. A coordinator that finds home replica D down writes to the next healthy node past the preference list — that stand-in is outside D's replica set, which is why deep dive 2's overlap guarantee weakens under failure. And a Merkle sweep descends only where subtree hashes differ, so it costs 13 comparisons to find one lost update among 200 keys.

The cluster class ties it together — put records hints, recover hands them off, anti_entropy repairs what the hints missed:

code/hld/kv_cluster.py — the cluster

class KVCluster:
    """Ring, quorum and versioning composed into one object.

    ``_ring`` decides where a key lives, ``_nodes`` hold the data. ``_lock`` guards every
    node's ``data``, ``hints`` and ``up`` flag; a real cluster locks per key on each node
    and talks over the network, but the quorum arithmetic is the same.
    """

    def __init__(
        self,
        nodes: Sequence[str],
        n: int = 3,
        w: int = 2,
        r: int = 2,
        vnodes: int = 64,
        sloppy: bool = True,
    ) -> None:
        if not nodes or len(set(nodes)) != len(nodes):
            raise ValidationError("nodes must be distinct and non-empty")
        if not 1 <= n <= len(nodes):
            raise ValidationError("need 1 <= N <= number of nodes")
        self.overlapping = quorum_overlaps(n, w, r)  # also validates 1 <= W, R <= N
        self.n, self.w, self.r, self.sloppy = n, w, r, sloppy
        self._ring = HashRing(nodes, vnodes=vnodes)
        self._nodes = {name: StorageNode(name) for name in nodes}
        self._lock = threading.RLock()

    def node(self, name: str) -> StorageNode:
        if name not in self._nodes:
            raise NotFoundError(f"no node named {name!r}")
        return self._nodes[name]

    def preference_list(self, key: str) -> list[str]:
        """The N distinct physical nodes clockwise from hash(key): the key's home replicas."""
        return self._ring.preference_list(key, self.n)

    def _targets(self, key: str) -> list[tuple[str, str]]:
        """``(home, holder)`` pairs: each home replica, or the next healthy node past the
        preference list standing in for a home that is down (a sloppy quorum)."""
        order = self._ring.preference_list(key, len(self._nodes))  # every node, ring order
        spare = (name for name in order[self.n :] if self._nodes[name].up)
        targets: list[tuple[str, str]] = []
        for home in order[: self.n]:
            if self._nodes[home].up:
                targets.append((home, home))
            elif self.sloppy and (stand_in := next(spare, None)) is not None:
                targets.append((home, stand_in))
        return targets

    def put(
        self, key: str, value: str | None, context: VectorClock | None = None, via: str | None = None
    ) -> WriteResult:
        """Write to the key's replicas through coordinator ``via`` (default: the first healthy
        one). Without a ``context`` the write is blind and becomes a sibling of anything
        concurrent; with the context of the last read it supersedes what that read saw."""
        with self._lock:
            targets = self._targets(key)
            holders = [holder for _, holder in targets]
            if len(holders) < self.w:
                raise QuorumError(f"write of {key!r}: only {len(holders)} replica(s) reachable, needed W={self.w}")
            coordinator = holders[0] if via is None else via
            if coordinator not in holders:
                raise ValidationError(f"{via!r} is not a healthy replica of {key!r}")
            local = self._nodes[coordinator].data.get(key, [])
            seen = max((v.clock.counter(coordinator) for v in local), default=0)
            version = Version(value, (context or VectorClock()).increment(coordinator, seen))
            hinted: list[tuple[str, str]] = []
            for home, holder in targets:
                node = self._nodes[holder]
                node.store(key, version)
                if holder != home:  # stand-in: serve reads now, hand the write back later
                    node.hints[(home, key)] = reconcile([*node.hints.get((home, key), []), version])
                    hinted.append((holder, home))
            return WriteResult(version.clock, tuple(holders), tuple(hinted))

    def delete(self, key: str, context: VectorClock | None = None) -> WriteResult:
        """A delete is the write of a tombstone: reads hide it, replication still ships it."""
        return self.put(key, None, context)

    def get(self, key: str) -> ReadResult:
        """Merge the siblings held by the first R healthy replicas and repair the stale ones."""
        with self._lock:
            answering = [holder for _, holder in self._targets(key)][: self.r]
            if len(answering) < self.r:
                raise QuorumError(f"read of {key!r} got {len(answering)} answers, needed R={self.r}")
            merged = reconcile(v for name in answering for v in self._nodes[name].data.get(key, []))
            repaired = tuple(name for name in answering if merged and self._nodes[name].store(key, *merged))
            context = VectorClock()
            for version in merged:
                context = context.merge(version.clock)
            values = tuple(v.value for v in merged if v.value is not None)
            return ReadResult(values, context, tuple(answering), repaired)

    def fail(self, name: str) -> None:
        with self._lock:
            self.node(name).up = False

    def recover(self, name: str) -> int:
        """Bring ``name`` back and deliver the hints other nodes kept for it (hinted handoff)."""
        with self._lock:
            node = self.node(name)
            node.up = True
            delivered = 0
            for other in self._nodes.values():
                for (home, key), versions in list(other.hints.items()):
                    if home != name:
                        continue
                    node.store(key, *versions)
                    del other.hints[(home, key)]
                    if other.name not in self.preference_list(key):
                        other.data.pop(key, None)  # the stand-in's copy has done its job
                    delivered += 1
            return delivered

    def anti_entropy(self, left: str, right: str, leaves: int = 64) -> SyncResult:
        """Compare the key range both nodes replicate via Merkle trees; sync only what differs."""
        with self._lock:
            a, b = self.node(left), self.node(right)
            shared = {k for k in (*a.data, *b.data) if {left, right} <= set(self.preference_list(k))}
            tree_a = MerkleTree(a.fingerprint(shared), leaves)
            tree_b = MerkleTree(b.fingerprint(shared), leaves)
            diff = tree_a.diff(tree_b)
            synced: list[str] = []
            for bucket in diff.buckets:
                for key in sorted({*tree_a.keys_in_bucket(bucket), *tree_b.keys_in_bucket(bucket)}):
                    merged = reconcile([*a.data.get(key, []), *b.data.get(key, [])])
                    a.store(key, *merged)
                    b.store(key, *merged)
                    synced.append(key)
            return SyncResult(diff.buckets, diff.comparisons, tuple(synced))

    def holders(self, key: str) -> list[str]:
        """Every node with a copy of ``key``, home replica or stand-in; for demos and tests."""
        with self._lock:
            return [name for name, node in self._nodes.items() if key in node.data]

Running python -m hld.kv_cluster prints the whole story:

5 nodes x 64 vnodes, N=3 W=2 R=2 (overlapping=True); cart:42 lives on ['C', 'A', 'D']
put 'apple', no context            -> clock {C:1}, stored on ['C', 'A', 'D']
get                                -> ['apple'] context {C:1}
two writes from {C:1} via A and D  -> clocks {A:1, C:1} and {C:1, D:1}
get                                -> siblings ['apple,bread', 'apple,milk'] context {A:1, C:1, D:1}
client merges, writes with context -> clock {A:1, C:2, D:1}; get -> ['apple,bread,milk']
C down, put                        -> acks ['E', 'A', 'D']; E keeps a hint for C
get while C is down                -> ['apple,bread,milk,eggs'] answered by ['E', 'A']
C recovers                         -> 1 hinted write handed off, copies on ['A', 'C', 'D']
strict quorum, C and A down        -> write of 'cart:42': only 1 replica(s) reachable, needed W=2
anti-entropy A vs D, lost hint     -> 1 bucket differs after 13 hash comparisons, compared ['item:164', 'item:7']; D now holds 'v7-updated'

Permanent removal is an operator action: the leaving node's successors stream its arcs from the remaining replicas.

Deep dive: the node write path from WAL to memtable to SSTable

The probing question is "what happens inside one replica when it receives store(key, version), and why is it fast?"

Engine	Write cost	Point read cost	Fits because
B-tree (update in place)	Random write per update, page splits	One tree descent	Read-heavy, range-heavy workloads
LSM tree (append, then merge)	Sequential append + deferred compaction	Memtable, then Bloom-filtered SSTables	Write-heavy, point-lookup workloads: this store
Hash index in memory + log	Append	One hash lookup	Dataset fits in RAM per node (Bitcask, Riak's default)

Inside one replica: append, buffer, flush, compact; reads skip tables via Bloom filters.

flowchart LR
    n_put["store(key, version)"] --> n_wal["WAL append + fsync"]
    n_wal --> n_ack["ack to coordinator"]
    n_wal --> n_mem["Memtable<br/>sorted, in memory"]
    n_mem -->|"memtable full"| n_flush["Flush as an immutable sorted run"]
    n_flush --> n_l0[("L0 SSTable<br/>+ Bloom filter + sparse index")]
    n_l0 -->|"compaction: merge, drop dominated versions and old tombstones"| n_ln[("L1..Ln SSTables")]
    n_get["read(key)"] --> n_mem
    n_get --> n_bloom{"Bloom filter says maybe?"}
    n_bloom -->|"no"| n_skip["skip the table"]
    n_bloom -->|"yes"| n_l0
    n_bloom -->|"yes"| n_ln

Whiteboard WAL -> memtable -> SSTable (Bloom + index) -> compaction and say "durability from the WAL, read speed from Bloom filters, write amplification as the bill". The two numbers to keep: a Bloom filter at ~10 bits per key rules out ~99% of the tables that lack the key, and the sparse index turns what remains into one block read (~16 µs of SSD). Everything else — skip-list memtables, immutable SSTables, compaction strategies and measured amplification — is on the storage engines page.

Scaling, bottlenecks and failure modes

v2: rack-aware replicas, a second region, and repair work isolated from the request path.

flowchart LR
    subgraph clients["Clients"]
        c_app["Apps with ring-aware client library"]
    end
    subgraph region_a["Region A"]
        a_coord["Coordinators: any node"]
        a_ring["Ring: 3 racks, rack-aware preference lists"]
        a_n1[("Rack 1 nodes")]
        a_n2[("Rack 2 nodes")]
        a_n3[("Rack 3 nodes")]
        a_hints[["Hint queues"]]
        a_repair["Repair scheduler: Merkle sweeps,<br/>throttled compaction"]
    end
    subgraph region_b["Region B"]
        b_coord["Local coordinators"]
        b_ring[("Full replica set, N=3 per region")]
    end
    subgraph ops["Operations"]
        o_gossip["Gossip + seeds"]
        o_backup[("Snapshots to object storage")]
    end
    c_app --> a_coord --> a_ring
    a_ring --> a_n1
    a_ring --> a_n2
    a_ring --> a_n3
    a_coord --> a_hints
    a_repair --> a_n1
    a_coord -->|"async cross-region replication"| b_coord --> b_ring
    o_gossip --> a_coord
    o_gossip --> b_coord
    a_n2 --> o_backup

What breaks first, and what you do about it:

• Hot keys. A viral key lands on exactly N nodes however large the cluster is: consistent hashing spreads keys, not popularity. Cache them, salt write-hot counters (key#0..key#9, merged on read), or move their tokens to quieter nodes.
• Tail latency. One replica in a long compaction or GC pause slows every quorum it joins; speculative retries after a p99 timeout and throttled compaction cover what waiting for R does not.
• Sibling, tombstone and hint growth. A client writing without a context outruns whoever merges siblings, so cap and alert on them. Deletes are writes, so a range full of them reads slowly until compaction purges tombstones; hints expire on a TTL, so a node down for days is repaired by anti-entropy instead of a flood on its return.
• Membership churn. A flapping node triggers hint storms, so gossip marks it suspect and leaves the ring alone until an operator decides; a joining node streams ~1/N of the data, throttled, and serves reads only once its ranges are complete. A split brain accepts writes on both sides — the AP promise, paid for in siblings.
• Cross-region. Replicate asynchronously with N=3 per region and local quorums (LOCAL_QUORUM): a ~70-150 ms cross-region hop inside every write would blow the 20 ms budget. Inter-region conflicts are ordinary siblings.

Trade-offs summary

Decision	Chosen	Alternatives	Why
Topology	Leaderless, any replica coordinates	Single leader per partition (Raft, Bigtable)	Always writable, no failover pause; the price is conflicts
Partitioning	Consistent hashing with virtual nodes	Range partitioning, hash mod N	~1/N keys move on a membership change
Replica placement	Preference list of N distinct, rack-aware nodes	Next N tokens	Two replicas on one rack is one failure, not two
Consistency knob	Tunable W and R, default 2/2 of 3	Fixed strong or fixed eventual	Overlap where it matters, speed where it does not
Versioning	Vector clocks with client-side merge; LWW opt-in	LWW everywhere	LWW silently loses concurrent updates and trusts clocks
Availability under failure	Sloppy quorum + hinted handoff	Strict quorum (refuse writes)	Writes survive an outage; the hint repays the debt
Repair	Read repair + scheduled Merkle anti-entropy	Full replica comparison	Cost scales with differences, not data size
Storage engine	LSM tree per node	B-tree	Append-only writes, Bloom-filtered point reads

Interviewer follow-ups

Why not pick a leader per partition and avoid conflicts entirely?

A leader gives a total order per key and no siblings, but every leader failure costs an election window during which the partition refuses writes. Dynamo needed cart writes never to fail.

What exactly goes wrong with last-writer-wins?

Two concurrent writes leave one survivor chosen by timestamp; the other vanishes silently. Clocks drift too, so a fast node wins every conflict until someone notices. Dangerous for carts and counters.

How do you keep vector clocks from growing without bound?

Entries are per coordinator, so a clock holds one per node that ever coordinated the key; Dynamo truncates to the ten newest. Truncation can fake a sibling, which the next reconciling write removes.

How does a new node join without downtime?

The operator adds it with a set of tokens, gossip spreads the new ring, and the node streams each range from the current replicas while they keep serving; it answers reads once streaming completes.

How would you add range queries or secondary indexes?

Range queries need order-preserving partitioning, trading even spread for hot ranges; Cassandra hashes the partition key and sorts on a clustering key within the partition. Secondary indexes go local or global.

Where do the Bloom filters live and what do they cost?

One per SSTable, in memory. At ~10 bits per key, the ~100M key copies each of the 30 nodes stores cost ~125 MB — trivial next to the row cache. A point read then skips almost every table.

Interview tip

Say "preference list" in your first sentence about replication and draw the ring with the N distinct nodes marked. It shows the replicas are not "the next N tokens", and it sets up sloppy quorums, hints and anti-entropy alike.

Common mistake

Claiming that W + R > N makes the store strongly consistent. It only makes a read see the latest acknowledged write under normal operation: concurrent writes still produce siblings, and sloppy quorums break the overlap.

45-minute pacing

Minutes	What to say and draw
0–5	Clarify: AP over CP, opaque ~1 KB values, 1B keys, 10:1 reads, single region, application merges conflicts.
5–9	Estimation: 100k reads/s, 10k writes/s, 3 TB raw, 90k replica writes/s peak, ~30 nodes — "sized by replica ops, not disk".
9–14	API (get returns siblings + context, put takes context, delete is a tombstone) and the per-node data model.
14–24	v1 diagram; narrate the write path (clock, WAL, W acks) and the read path (R answers, reconcile, repair).
24–40	Deep dives: ring and preference lists, N/W/R overlap, clocks vs LWW, the failure hour; LSM path if asked.
40–45	Bottlenecks (hot keys, tail latency, sibling and tombstone growth), multi-region quorums, trade-offs.

Related

• Partitioning, sharding and consistent hashing — the HashRing the cluster is built on
• Replication — quorum arithmetic, the stale-read demo, Merkle trees
• Time, clocks and ordering — why wall clocks cannot order concurrent writes
• Storage engines and indexing — the LSM tree inside each node
• CAP, PACELC and consistency models — where "always writable" sits
• Classic papers digest — Dynamo and Cassandra in a page each
• Primary source: DeCandia et al., "Dynamo: Amazon's Highly Available Key-value Store" (SOSP 2007)