system-design · intermediate

Distributed Caching — Sharding and High-Availability Clusters

The Central Question

Consider a global platform running on the CacheLab platform (cachelab.com) processing 50,000 requests per second:

• The application tier requires 500 GB of in-memory caching capacity to store user sessions, product catalogs, and rate-limiting counters.
• A single high-end server node provides 64 GB of RAM.
• A single Redis node processes up to 100,000 operations per second on its single-threaded event loop. Total application traffic requires 1,500,000 operations per second.

A Distributed Cache pools the memory and CPU capacity of multiple dedicated cache server nodes into a single, cohesive, highly available in-memory data cluster.

This lesson answers one central question: How do distributed cache architectures (Redis Cluster, Memcached) partition key spaces using Consistent Hashing, maintain cluster availability using primary-replica failover, and prevent Hot Key node saturation?

Standalone Cache vs. Distributed Cache Cluster

As data volume and QPS grow beyond single-server limits, architectures evolve from standalone nodes to distributed clusters:

flowchart TB
  subgraph Standalone Cache Limit
    App1[App Pods] -->|Single Node Bottle-Neck| SingleRedis[(Standalone Redis Node<br/>Max RAM: 64 GB | Max QPS: 100k)]
  end
subgraph Distributed Cache Cluster (Redis Cluster)
    App2[App Pods] --> Router{Hash Slot Router: CRC16 key % 16384}
    Router -->|Slots 0 - 5460| NodeA[(Cache Node A: 64 GB RAM)]
    Router -->|Slots 5461 - 10922| NodeB[(Cache Node B: 64 GB RAM)]
    Router -->|Slots 10923 - 16383| NodeC[(Cache Node C: 64 GB RAM)]
  end

Figure 1: Transition from single-node bottleneck to 3-node sharded distributed cache cluster.

Partitioning Key Spaces: Consistent Hashing & Redis Hash Slots

A distributed cache must map arbitrary string keys (e.g. session:usr_9021) to specific physical cache server nodes.

1. Modulo Hashing Anti-Pattern

$$\text{Target Node} = \text{hash}(\text{key}) \pmod N$$

Where $N$ is the number of cache nodes in the cluster.

The Disaster of Modulo Hashing: If Node 3 fails or a new Node 4 is added ($N$ changes from 3 to 4), nearly 100% of all keys map to different nodes. The entire cluster suffers a catastrophic, simultaneous cache miss, triggering a massive database outage.

2. Redis Cluster Hash Slots (16,384 Fixed Slots)

$$\text{Slot} = \text{CRC16}(\text{key}) \pmod{16384}$$

flowchart TD
  Key["Key: 'user:session:9021'"] --> CRC["CRC16('user:session:9021') = 45,920"]
  CRC --> Mod["45920 % 16384 = Hash Slot 13184"]
  Mod --> SlotMap{Cluster Slot Allocation Table}
  
  SlotMap -->|Slot 13184| NodeC["Node C (Manages Slots 10923 - 16383)"]

Figure 2: Redis Cluster key slot calculation routing a request to Node C.

When a new node joins a Redis Cluster, only a fraction of the 16,384 slots are migrated from existing nodes. Over 90% of cached keys remain unaffected in their original slot assignments.

Consistent Hashing with Virtual Nodes

For client-side distributed caching (such as Memcached), Consistent Hashing maps both server nodes and cache keys onto a continuous $2^{32}-1$ integer ring:

flowchart TD
  subgraph Consistent Hashing Ring
    Ring["360 Degree Hash Ring (0 to 2^32 - 1)"]
    NodeA["Node A_v1 (Angle 30°)"]
    NodeB["Node B_v1 (Angle 150°)"]
    NodeC["Node C_v1 (Angle 270°)"]
    
    Key1["Key 'user_88' (Angle 85°) -> Routes Clockwise to Node B"]
  end

Figure 3: Consistent Hashing ring placement routing keys clockwise to virtual nodes.

Virtual Nodes for Uniform Load Distribution

Catastrophic Distributed Cache Failures: Avalanche, Penetration, and Stampede

Operating a large distributed cache cluster introduces three classic distributed system failure modes:

flowchart TD
  FailureModes[Distributed Cache Failure Modes] --> Avalanche[1. Cache Avalanche]
  FailureModes --> Penetration[2. Cache Penetration]
  FailureModes --> Stampede[3. Cache Stampede]
  
  Avalanche --> AvDesc["Thousands of keys expire simultaneously.<br/>Database flooded with concurrent queries.<br/>Fix: Add Random TTL Jitter!"]
  Penetration --> PenDesc["Queries request non-existent keys (id=-1).<br/>Cache misses every time; slams DB.<br/>Fix: Deploy Bloom Filters!"]
  Stampede --> StaDesc["Hot key expires; 5,000 threads recompute at once.<br/>DB CPU hits 100%.<br/>Fix: Probabilistic Early Expiration (XFetch)!"]

Figure 4: Taxonomy of Cache Avalanche, Penetration, and Stampede failure modes.

1. Cache Avalanche

• Cause: Thousands of hot keys are written with identical TTL durations (e.g. 3,600 seconds) during a midnight batch update, causing all keys to expire simultaneously.
• Mitigation: Add Random TTL Jitter ($\text{TTL} = 3600 \pm \text{rand}(0, 300)$ seconds) to stagger expirations smoothly over time.

2. Cache Penetration

• Cause: Malicious users or broken clients query non-existent keys (e.g. user_id = -9999). The cache returns Miss, forcing every single request to query the underlying database.
• Mitigation: Deploy a Bloom Filter at the gateway to check if the key exists before querying the cache, or store explicit Null Values with Short TTLs in the cache (SET key NULL EX 60).

3. Cache Stampede (Thundering Herd)

• Cause: A single ultra-hot key (e.g. product_iphone_sale receiving 10,000 QPS) expires. Thousands of application threads experience a cache miss simultaneously and attempt to re-compute the expensive SQL join concurrently.
• Mitigation: Implement Probabilistic Early Expiration (XFetch Algorithm) or Distributed Locks (Redlock) to allow only 1 thread to recompute the key while others return stale data.

Hot Key Scattering Techniques

1. When writing a hot key, replicate the payload across $M$ distinct keys using random integer suffixes (hot_key#1, hot_key#2, ..., hot_key#10).
2. Each suffix maps to a different hash slot and node in the Redis Cluster.
3. Read clients pick a random suffix rand(1, 10) to distribute the 50,000 QPS evenly across 10 distinct physical server nodes.

Managing Redis Memory Fragmentation

High Availability via Redis Sentinel

Complete Worked Example: Go Consistent Hashing Ring with Virtual Nodes

Let's inspect a complete Go implementation of a Consistent Hashing ring with virtual nodes for the CacheLab platform (cachelab.com).

package main
import (
	"fmt"
	"hash/fnv"
	"sort"
	"strconv"
	"sync"
)
type HashRing struct {
	mu       sync.RWMutex
	vNodes   int            // Virtual nodes per physical node
	ring     []uint32       // Sorted list of hash positions
	nodeMap  map[uint32]string // Hash position -> Physical Node ID
}
func NewHashRing(vNodes int) *HashRing {
	return &HashRing{
		vNodes:  vNodes,
		nodeMap: make(map[uint32]string),
	}
}
func (h *HashRing) hash(key string) uint32 {
	hasher := fnv.New32a()
	hasher.Write([]byte(key))
	return hasher.Sum32()
}
func (h *HashRing) AddNode(nodeID string) {
	h.mu.Lock()
	defer h.mu.Unlock()
for i := 0; i < h.vNodes; i++ {
		vNodeKey := nodeID + "#vnode" + strconv.Itoa(i)
		hashPos := h.hash(vNodeKey)
		h.ring = append(h.ring, hashPos)
		h.nodeMap[hashPos] = nodeID
	}
sort.Slice(h.ring, func(i, j int) bool { return h.ring[i] < h.ring[j] })
}
func (h *HashRing) GetNode(key string) string {
	h.mu.RLock()
	defer h.mu.RUnlock()
if len(h.ring) == 0 {
		return ""
	}
hashPos := h.hash(key)
// Binary search for closest node clockwise on ring
	idx := sort.Search(len(h.ring), func(i int) bool { return h.ring[i] >= hashPos })
	if idx == len(h.ring) {
		idx = 0 // Wrap around to start of ring
	}
return h.nodeMap[h.ring[idx]]
}

Failure Modes and Engineering Mitigations

What You Should Remember

1. Distributed Caches scale RAM and QPS beyond single-node limits: Pool multiple nodes to store terabytes of data and process millions of QPS.
2. Use Consistent Hashing & Hash Slots: Prevent mass cache invalidation during node additions or failures by mapping keys to 16,384 fixed slots or hash rings.
3. Prevent Cache Avalanche with Random TTL Jitter: Stagger key expirations to prevent simultaneous mass cache misses from crashing databases.
4. Deploy Bloom Filters against Cache Penetration: Block invalid queries for non-existent keys before they reach the cache or database.
5. Mitigate Hot Keys via Key Scattering: Scatter viral keys across multiple nodes (key_1, key_2) to distribute CPU load evenly across cluster nodes.

Glossary of Terms

Practice Scenario and Self-Assessment

Architecture Scenario

• A single leaderboard key (leaderboard_global) receives 80,000 QPS, causing a single Redis node CPU to hit 100%.

1. Formulate a Key Scattering strategy to distribute leaderboard_global reads across 10 Redis Cluster nodes.
2. Calculate the slot assignment and virtual node distribution required to maintain uniform load across the cluster.

Interactive Self-Assessment

Changing N from 3 to 4 nodes re-maps nearly 100% of all keys, causing a global simultaneous cache miss that crashes the database.

Modulo hashing causes physical electrical short-circuits in server NIC network interfaces.

Modulo hashing converts relational database SQL schemas into flat CSV files.

Modulo hashing automatically updates client browser local hosts file records.

It staggers key expirations smoothly over time, preventing thousands of keys from expiring simultaneously and flooding the database.

TTL Jitter automatically re-generates edge CDN HTTPS TLS encryption certificates.

TTL Jitter formats the underlying server persistent NVMe SSD disk drives.

TTL Jitter reboots the operating system hypervisor on secondary backup nodes.

What to Learn Next

• Content Delivery Networks — CDNs & Edge Caching: Learn edge proxy caching, HTTP headers, and Anycast PoPs.
• Cache Eviction Policies — LRU, LFU, FIFO, ARC: Revisit $O(1)$ LRU and W-TinyLFU eviction algorithms.
• Caching Strategies — Read-Through, Write-Through, Write-Back: Review Write-Behind batching patterns.

Track: Data, Storage and Messaging

Previous: Distributed Cache Design

Next: Read-Through vs. Write-Through Cache — Who Updates the Cache?

Series: Caching

1. Caching 101 — Memory Offloading and Latency Reduction
2. Caching Strategies — Aside, Through, Behind, and Refresh-Ahead
3. Cache Eviction Policies — LRU, LFU, TTL, and Friends
4. Distributed Caching — Sharding and High-Availability Clusters (this guide)
5. Cache Stampede — When Expiry Melts the Database
6. Stale Cache After Write — When Your Own Update Disappears
7. Content Delivery Networks (CDN) — Edge Acceleration and Caching
8. Bloom Filters — Probabilistic Set Membership at Scale

All series

By Shubham Jain

All articles · Study paths

Shubham Jain · Learning Lab