02. Distributed Cache — Low-Level Design¶

~22 min read · Part 3 of 4 (Overview → HLD → LLD → Q&A)

The HLD named the boxes. This file opens the four that carry the design's weight — the consistent-hash ring, replication and failover, the eviction engine, and the hot-key / stampede defenses — and pins down the data structures, the algorithms, and the concurrency corners where a cache actually breaks. The threaded scenario (100 nodes, one dies at peak, 1% of keys move, origin stays flat) runs through the ring algorithm and the failover sequence.

Data models¶

A cache node's core is one in-memory hash table, but each entry carries more than a value so eviction and expiry can reason about it:

CacheEntry {
    key        : bytes            # the lookup key
    value      : bytes            # opaque payload, ~2 KB average here
    expire_at  : uint64           # absolute ms; 0 = no TTL
    last_access: uint32           # coarse timestamp, for LRU
    freq       : uint8            # LFU access counter (probabilistic, see below)
    lru_prev, lru_next : *Entry   # intrusive doubly-linked list pointers
}

Two deliberate choices. freq is a single byte, not a real counter — an exact per-key hit count would cost more memory than many values and would never stop growing; the LFU section explains the logarithmic-increment-plus-decay scheme that makes 8 bits enough. And the LRU pointers are intrusive (stored in the entry itself, not a side structure) so that recording an access is O(1) pointer surgery with no allocation, which matters when every one of 100k ops/second/node touches them.

The ring lives in the control plane and is cached by every client:

RingEntry {
    token   : uint64      # position on the 2^64 hash ring
    node_id : string      # physical node that owns this virtual point
}
Ring = sorted list of RingEntry by token   # 100 nodes × 160 vnodes = 16,000 entries

SlotMap {
    slot -> { primary: node_id, replicas: [node_id, ...] }   # Redis-style: 16,384 slots
}

The ring is stored as a sorted array of tokens, not a tree, because it is read on every routed request (binary search, O(log V)) and written only on rare topology changes — optimize for the read. Redis Cluster collapses the continuous ring into 16,384 fixed slots so the map is a small fixed-size array and moving keys means moving whole slots; the two representations are the same idea at different granularities.

Replica bookkeeping on each primary:

ReplicationState {
    replica_offset : uint64   # bytes of write-stream the replica has acked
    primary_offset : uint64   # bytes the primary has produced
    # lag = primary_offset - replica_offset ; bounds write loss on failover
}

Component internals¶

Component 1 — The consistent-hash ring¶

Responsibility: map any key to its owning node such that adding or removing a node moves only that node's share of keys.

class Ring:
    tokens: sorted[uint64]                 # 16,000 virtual-node positions
    owner : map[token -> node_id]

    def route(key) -> node_id:
        h = hash64(key)                    # e.g. xxhash / murmur, uniform
        i = lower_bound(tokens, h)         # first token >= h (binary search)
        if i == len(tokens): i = 0         # wrap around the ring
        return owner[tokens[i]]

    def add_node(node_id):
        for v in range(VNODES_PER_NODE):   # 160 virtual points
            t = hash64(node_id + "#" + v)
            insert(tokens, t); owner[t] = node_id

    def remove_node(node_id):
        drop every token whose owner == node_id     # ~160 arcs, each reassigned
                                                     # to the next clockwise token's owner

The reason virtual nodes exist is load smoothing. With one point per physical node, a node's arc is however wide the random gap to its neighbor happens to be — some nodes get 3× the keys of others, and when a node dies its entire arc dumps onto one neighbor. With 160 points per node, each node's share is the sum of 160 small random arcs, which by the law of large numbers is close to 1/100 for every node, and a dead node's 160 arcs each hand off to a different clockwise neighbor — so its 1% of keys spreads across ~99 survivors, not onto one.

Component 2 — Replication and failover¶

Responsibility: keep a warm copy of every primary's keys, and promote it on primary death without splitting ownership.

# primary side, per write
def on_write(entry):
    apply_local(entry)                     # in-memory, atomic (single-threaded shard)
    repl_stream.append(encode(entry))      # async; does NOT block the client ack
    primary_offset += len(encode(entry))

# replica side
def on_repl_batch(batch):
    for entry in batch: apply_local(entry)
    ack(replica_offset = primary_offset_seen)

# coordinator, on suspected death (gossip: node unreachable by a quorum)
def failover(dead_primary):
    if not quorum_agrees_dead(dead_primary): return    # guard split-brain
    replica = best_replica(dead_primary)               # highest replica_offset
    promote(replica)                                   # replica becomes primary
    publish_slotmap()                                  # clients refresh routing

Two guards matter. quorum_agrees_dead requires a majority of primaries to vote a node dead before any promotion, so a brief network blip or a partition minority cannot trigger a spurious failover that creates two primaries for one slot. And best_replica picks the replica with the highest replica_offset (least lag) so failover loses the fewest writes.

Component 3 — The eviction engine (LRU and LFU)¶

Responsibility: when memory exceeds budget, remove the entries least worth keeping, in O(1) amortized time, without scanning the whole keyspace.

Exact LRU keeps the intrusive doubly-linked list ordered by recency: on every access, unlink the entry and move it to the head; to evict, drop the tail. O(1), but the pointer writes on every read add up, and it is vulnerable to scans.

Redis instead uses approximate LRU/LFU via sampling to avoid maintaining a global order at all:

def evict_one(policy):
    sample = random_keys(count=5)          # sample, don't scan
    if policy == LRU:
        victim = min(sample, key=lambda e: e.last_access)
    elif policy == LFU:
        victim = min(sample, key=lambda e: e.freq)
    delete(victim)

For LFU, the freq byte cannot be a true count — a hot key would overflow 8 bits in milliseconds. It is a logarithmic, probabilistic counter with decay:

def touch_lfu(entry):
    # increment with probability that shrinks as freq grows:
    if random() < 1.0 / (entry.freq * LFU_LOG_FACTOR + 1):
        entry.freq = min(entry.freq + 1, 255)
    # separately, decay freq over time so once-hot keys can fall out:
    entry.freq = max(entry.freq - elapsed_minutes / LFU_DECAY, base_floor)

The probabilistic increment means freq grows roughly with the logarithm of access count, so 8 bits distinguish "hit ten times" from "hit ten million times." The decay term is what lets a formerly hot key age out after it stops being accessed — without it, LFU would pin yesterday's hot keys forever and never adapt.

Core algorithm — node loss at peak, on the ring¶

Walk the threaded scenario through the ring math step by step.

Setup. 100 nodes, each with 160 virtual points → 16,000 tokens on a 2^64 ring. Keys hash uniformly, so each physical node owns ~1/100 = 1% of the 2^64 space, split into 160 small arcs. At 10M ops/second, node 42 is serving ~10M / 100 = 100,000 ops/second across its ~5M keys (500M keys / 100).

Step 1 — node 42 dies. Gossip detects it unreachable; a quorum of primaries votes it dead. Only node 42's 160 tokens are removed from the ring. Every key whose hash falls in one of those 160 arcs now resolves, via route(), to the next clockwise token — which belongs to some other physical node. Every key not in those arcs finds the same owner as before, because its clockwise-nearest surviving token is unchanged. That is the crux: ~1% of keys change owner, ~99% do not. Compare with hash(key) % N: going from % 100 to % 99 would relocate ~99% of keys — the exact inversion of what we want.

Step 2 — the 1% redistributes evenly. Node 42's 160 arcs each hand off to whatever node owns the next token clockwise — 160 different handoff targets, statistically spread across the 99 survivors. So the departing 100,000 ops/second splits into ~100,000 / 99 ≈ 1,010 extra ops/second per surviving node, lifting each from 100k to ~101k ops/second — about a 1% bump, invisible on the cache side. Without virtual nodes, node 42's single arc would have dumped all 100,000 ops/second onto one neighbor, doubling its load and likely toppling it — a cascading failure. Virtual nodes are what convert a cliff into a ramp.

Step 3 — keep the origin flat. The 1% of keys just changed owner, and on their new owner they would be cold — the naive outcome is 100,000 ops/second of misses hitting a database already at its 100k baseline, driving it to 200k past its 150k ceiling. Two mechanisms prevent it. First, warm replica promotion: node 42's replica (say node 43 held its slots) already has those 5M keys in memory, so the coordinator promotes node 43 and the "moved" keys are hits from the very first request — origin QPS does not move at all. Second, as a backstop for the tiny window of keys written to node 42 but not yet replicated when it died, single-flight coalescing on the client ensures that if 5,000 concurrent requests all miss on the same just-lost key, exactly one reads origin and the other 4,999 wait for its result. So even in the degraded no-replica case, origin sees at most one read per distinct cold key, not one per request. That is the precise mechanism behind "no stampede reaches origin."

Sequence diagram — failover with a warm replica, no origin hit¶

Client        Coordinator     Node42(primary)   Node43(replica)   Origin
  │   GET k (k owned by 42)        │                │               │
  ├───────────────────────────────▶│  ✗ (dead)      │               │
  │                                 X                │               │
  │        (gossip: 42 unreachable) │                │               │
  │◀─ quorum votes 42 dead ─────────┤                │               │
  │                                 │  promote(43) ─▶│               │
  │◀─ new slotmap (k → 43) ─────────┤                │               │
  │   GET k                                          │               │
  ├─────────────────────────────────────────────────▶│  hit (warm)  │
  │◀──────────────────────── value ──────────────────┤               │
  │                                                  │   (origin never queried)

The replica already held k, so the retried GET after the routing refresh is a hit — the database is never touched. Had k been unreplicated, the same retry would miss, and single-flight would let exactly one client fetch it from origin while the rest waited.

Concurrency and edge cases¶

Concurrent SET on the same key: resolved by the shard being single-threaded — writes serialize with no lock, so last-writer-wins is well-defined and there is no torn value. For read-modify-write, callers must use CAS (below) rather than GET then SET.
Read-modify-write races (CAS): a GET-then-SET from two clients can lose an update — both read 10, both write 11, one increment vanishes. CAS key old new succeeds only if the current value still equals old; the loser retries. INCR is the atomic special case, done entirely inside the single-threaded shard so it never races.
Stale write after invalidation (the look-aside race): client A reads a miss and fetches v1 from origin; before A writes v1 into cache, client B updates origin to v2 and deletes the (absent) cache key; A then writes the now-stale v1, which lingers until TTL. Mitigations: short TTLs bound the staleness, delete-on-write plus a small delay, or a versioned CAS-populate that only fills if no newer version is present. This is why a cache is weakly consistent by design — the fix is a bounded-staleness contract, not an attempt at perfect coherence.
Thundering herd on TTL expiry: a hot key's TTL fires and thousands of reads miss simultaneously. Beyond single-flight, use probabilistic early recomputation — each read of a near-expiry hot key refreshes it with a probability that rises as expiry approaches, so one lucky early reader refreshes it before it ever actually expires, and the herd never forms. Serving the stale value during that refresh keeps latency flat.
Hot-key saturation: one key at 1M req/second saturates its owner's single thread regardless of cluster size. Detected by per-key op counters on the node; mitigated by the client L1 cache and by key-splitting (k#0..k#7 across 8 nodes). Consistent hashing does not and cannot help — it distributes distinct keys, and this is one key.
Split-brain on partition: if node 42 is merely partitioned, not dead, both it (on the minority side) and its promoted replica (on the majority side) could accept writes for the same slot. The quorum rule fixes ownership: only the majority side may serve the contested slot; node 42, finding itself unable to reach a quorum, stops serving rather than diverge. Availability on the minority is sacrificed to keep ownership single-valued.
Replica lag window: the primary_offset - replica_offset gap is the exact set of writes a failover can lose. Monitoring it turns durability risk into a number; on a clean (planned) failover the primary drains the stream to zero lag first, so only unplanned death loses writes, and only the sub-second tail.