Design a Distributed Cache — Consistent Hashing, Eviction, and Replication

"Design a distributed cache" is really two interviews stacked on top of each other. First: build an in-memory cache on one machine — which is the classic O(1) LRU question. Then the twist that makes it distributed: the hot data is bigger than one machine's RAM, so you spread it across a cluster — and now you have to answer how keys map to nodes without every key moving when a node joins, what happens when a node dies, and how the cache stays consistent with the database behind it. Get consistent hashing and the write policies right and you've got it.

A distributed cache is an in-memory key-value store spread across many nodes, sitting between your app and a slower backing store (a database). It exists to do one thing: serve reads in sub-millisecond time so the database doesn't have to. The interesting parts aren't get/put — they're how you shard keys across nodes (consistent hashing), what you evict when memory fills (LRU/LFU), and how you keep the cache and database in sync (write policies + invalidation). (New here? Start with the framework.)

TL;DR — the design at a glance

Concern	Decision
Single-node cache	Hashmap + doubly-linked list → O(1) LRU get/put
Sharding	Consistent hashing with virtual nodes
Eviction	LRU default; LFU for skewed popularity
Availability	Replication (primary + replicas), promote on failure
Cache ↔ DB	Cache-aside by default; write-through / write-back / write-around as needed
Consistency	Eventually consistent — accept a small stale window

• It's two problems: an O(1) single-node cache, and distributing it across a cluster.
• Never shard with hash(key) % N — changing N remaps almost every key and stampedes your DB. Use consistent hashing so only ~1/N of keys move when a node joins or leaves.
• Virtual nodes spread each physical node across many ring positions → balanced load, no hotspots from uneven placement.
• LRU (recency) is the default; LFU (frequency) wins when a few keys are persistently popular.
• Replicate shards for availability and read scale; promote a replica when a primary dies.
• Cache-aside is the default read/write pattern; know write-through (fresh, slower) and write-back (fast, lossy) too.
• Cache invalidation and the thundering herd are the genuinely hard parts — TTL + jitter + single-flight, not wishful thinking.
• A cache is eventually consistent with the DB — there's always a stale window after a write. Design for it or don't cache.

The one mental shift: a cache is a performance optimization that's allowed to be wrong sometimes. That single permission — eventual consistency, occasional misses, evicted data — is what lets it be fast and simple. Every hard decision (eviction, invalidation, write policy) is really "how wrong, for how long, is acceptable?"

Step 1 — Requirements & scope

Functional:

• get(key), put(key, value, ttl), delete(key).
• In-memory key-value store with per-key TTL / expiry.

Parked: rich data structures (that's Redis-specific), cross-datacenter replication, persistence to disk (it's a cache — the DB is the source of truth).

Non-functional:

• Sub-millisecond latency — the whole reason it exists.
• High throughput — millions of ops/sec.
• Horizontally scalable — the hot dataset is bigger than one machine's RAM.
• Highly available — a node dying should lose at most its shard, not the cache.
• Eventually consistent — acceptable, and what keeps it fast.

Step 2 — Back-of-envelope estimates

Say we front a service doing 1,000,000 reads/sec with a target 90%+ hit rate, over ~1 TB of hot data.

1 TB hot data ÷ ~100 GB RAM/node  ≈  10-20 cache nodes   (data won't fit one box)
1M reads/sec × 90% hit            ≈  900k served from cache, ~100k fall through to the DB

Two conclusions: the data must be sharded across a cluster (Step 6), and even a 90% hit rate means the cache absorbs the overwhelming majority of load — which is the point.

Step 3 — API

get(key)              -> value | MISS
put(key, value, ttl)  -> ok
delete(key)           -> ok

The client library (or a proxy) is responsible for routing a key to the node that owns it — the routing logic is where the design lives.

Step 4 — The single-node cache: O(1) LRU

Before distributing anything, build the one-machine cache. The classic design gives O(1) get and put with LRU eviction using two structures:

• A hashmap key → node for O(1) lookup.
• A doubly-linked list ordered by recency: most-recently-used at the head, least at the tail.

get(key):   look up in map; move its node to the head; return value
put(key,v): insert at head; if over capacity, evict the tail (LRU); update map

Moving a node to the head and evicting the tail are both O(1) with a doubly-linked list. This little structure is the atom every cache node is built from.

Step 5 — High-level architecture

   App / client library
        |  route key -> owning node
        v
   [ Cache cluster ]
     node A (shard + replicas)
     node B (shard + replicas)
     node C (shard + replicas)
        |  miss
        v
   [ Database ]  (source of truth)

Each node owns a shard of the keyspace plus one or more replicas of other shards. The client hashes the key to find its owner. The only question left — and the crux of the whole design — is how keys map to nodes.

Step 6 — Deep-dive: consistent hashing

Why hash(key) % N fails

The obvious sharding is node = hash(key) % N. It works until N changes. Add one node (N → N+1) and the modulus changes for almost every key, so nearly the entire cache remaps to different nodes — a mass of misses that all fall through to the database at once. Adding capacity shouldn't nuke your cache.

Consistent hashing

Map both nodes and keys onto a ring — a hash space of 0 … 2^32. A key is owned by the first node clockwise from the key's hash.

  Ring (clockwise):
    0 ──── A ──── key1 ──── B ──── key2 ──── C ──── key3 ──(wraps)── 0
                    └▶ B          └▶ C          └▶ A (wraps around)

  A key belongs to the first node clockwise from its position.

The magic is what happens on a change: add node D between B and C, and only the keys in the arc (B … D] move to D — roughly 1/N of the keys. Everything else stays put. Removing a node likewise only reassigns that node's arc to its clockwise neighbor. Adding or losing a node costs ~1/N churn instead of ~everything.

Virtual nodes

One problem remains: with only N points on the ring, placement is uneven — one node can own a much bigger arc than another, creating hotspots. Virtual nodes fix it: place each physical node at many points on the ring (e.g., 100–200 virtual positions). Load averages out across all those points, big and small nodes can be weighted by how many vnodes they get, and when a node leaves, its load spreads evenly across all others instead of dumping onto one neighbor.

Eviction policies

When a node's memory fills, it evicts:

• LRU (least recently used) — evict what hasn't been touched longest. The default; great for temporal locality.
• LFU (least frequently used) — evict the least-often used. Better when a stable set of keys is persistently hot and you don't want a one-off scan to flush them.
• FIFO / Random / TTL-based — simpler, occasionally useful; TTL expiry runs alongside whichever policy you pick.

Step 7 — Replication, write policies & the hard parts

Replication & availability

Each shard has a primary + one or more replicas. Replicas serve reads (scaling read throughput) and stand ready for failover — when a primary dies, a replica is promoted and only that shard briefly degrades. Without replication, a dead node means every key it held becomes a miss.

Keeping the cache and database in sync — write policies

This is where cache correctness lives:

• Cache-aside (lazy loading) — the default. App reads the cache; on a miss it reads the DB and populates the cache. On a write, it updates the DB and invalidates (deletes) the key. Simple and resilient; the trade is a stale window and a miss-penalty on cold keys.
• Write-through — write to cache and DB synchronously. The cache is always fresh, at the cost of higher write latency.
• Write-back (write-behind) — write to cache, flush to DB asynchronously. Fast writes, but a crash before the flush loses data — only for data you can afford to lose.
• Write-around — write straight to the DB, skip the cache; it fills on later reads. Good for write-once-read-rarely data you don't want polluting the cache.

The genuinely hard parts

• Cache invalidation — keeping the cache from serving stale data after a DB write. TTL is the pragmatic default; explicit delete-on-write is tighter; event-driven invalidation is tightest and most complex.
• Thundering herd (cache stampede) — when a hot key expires, thousands of simultaneous misses hammer the DB at once. Fixes: single-flight/locking (only one request recomputes, the rest wait), jittered TTLs (so keys don't all expire together), and stale-while-revalidate (serve the stale value while one request refreshes it).
• Hot keys — one wildly popular key overwhelms its single owning node; consistent hashing can't split a single key. Mitigate by replicating the hot key to multiple nodes or caching it locally in the client.

Step 8 — Trade-offs & wrap-up

• Consistent hashing vs modulo: ~1/N churn vs near-total remap. Always consistent hashing at scale.
• LRU vs LFU: recency vs frequency — pick from your access pattern.
• Write policy: cache-aside (simple, stale window) vs write-through (fresh, slower) vs write-back (fast, lossy) vs write-around (avoids pollution).
• Consistency: the cache is eventually consistent with the DB — there's always a window where it's stale. If you can't tolerate that, a cache may be the wrong tool.
• Memcached vs Redis: Memcached is a simple, multithreaded, pure cache; Redis adds rich data structures, replication, and optional persistence. Both are what you'd actually deploy instead of building this.

The design checklist

• Single-node cache: hashmap + doubly-linked list = O(1) LRU.
• Sharding via consistent hashing — never hash % N.
• Virtual nodes for balanced load.
• Eviction policy chosen (LRU default, LFU for skew) + TTL.
• Replication with failover promotion.
• A write policy chosen and justified (cache-aside by default).
• Planned for invalidation, thundering herd, and hot keys.
• Stated the eventual-consistency window.

In production at Mattrx

Mattrx is a multi-tenant marketing-analytics SaaS (110k MAU, ~3,200 req/sec peak) whose dashboard recomputes campaign KPIs — impressions, clicks, conversions, spend — by aggregating over a 1.2B-row CampaignEvents table in Azure SQL. Every dashboard load fanned out into repeated full aggregate scans, so the same handful of tenant/campaign/date-range combinations were being recomputed thousands of times a minute; the p95 for those queries sat at 2,100ms and pinned the database. Dropping the cache-aside layer above in front of those aggregates — short TTL for freshness, version-token invalidation on ingestion for correctness — moved almost all reads off the database and onto Redis.

Metric	Before	After
Dashboard KPI query p95	2,100ms	48ms
Database CPU	78%	22%
Working set	2.1GB	380MB
SQL cost	baseline	~$280/mo saved

The database went from absorbing every dashboard render to serving only genuine cache misses and fresh ingestion, which is what turned a 2,100ms tail into a 48ms one without ever handing users stale numbers.

The architecture at Mattrx

  React dashboard ──▶ [ .NET 9 KPI service (CQRS) ]
                            │  cache-aside
                            ▼
                     [ Azure Cache for Redis ]   key = tenant:campaign:range
                            │ miss
                            ▼
                     [ Azure SQL · CampaignEvents (1.2B rows) ]
                            ▲
       Kafka ingestion ─────┘  invalidates affected keys on new events

Production implementation (Mattrx)

Here is the cache-aside KPI service that sits in front of aggregate queries over Mattrx's 1.2B-row CampaignEvents table: a get-or-compute over IDistributedCache (Redis), keyed by tenant:campaign:dateRange, with a short TTL and event-driven invalidation triggered by Kafka ingestion.

using System.Text.Json;
using System.Text.Json.Serialization;
using Microsoft.Extensions.Caching.Distributed;
using Microsoft.Extensions.Logging;

namespace Mattrx.Analytics.Application.Dashboards;

public sealed record DateRange(DateOnly From, DateOnly To)
{
    // Stable, collision-free segment for the cache key.
    public override string ToString() => $"{From:yyyyMMdd}-{To:yyyyMMdd}";
}

public sealed record CampaignKpis(
    long Impressions,
    long Clicks,
    long Conversions,
    decimal Spend,
    double ConversionRate,
    DateTimeOffset ComputedAt);

// Aggregation read model over the 1.2B-row CampaignEvents table
// (Azure SQL, partitioned + indexed by tenant). One call = one expensive scan.
public interface IKpiQueryRepository
{
    Task<CampaignKpis> AggregateAsync(
        Guid tenantId, Guid campaignId, DateRange range, CancellationToken ct);
}

// Source-generated serialization: no reflection on the hot path.
[JsonSerializable(typeof(CampaignKpis))]
internal sealed partial class KpiJsonContext : JsonSerializerContext;

public sealed class DashboardKpiService(
    IDistributedCache cache,
    IKpiQueryRepository repository,
    TimeProvider clock,
    ILogger<DashboardKpiService> logger)
{
    // Short TTL is a backstop, not the correctness mechanism: ingestion-driven
    // invalidation bumps the version below, so 60s only bounds worst-case staleness.
    private static readonly DistributedCacheEntryOptions DataEntry = new()
    {
        AbsoluteExpirationRelativeToNow = TimeSpan.FromSeconds(60)
    };

    private static readonly DistributedCacheEntryOptions VersionEntry = new()
    {
        AbsoluteExpirationRelativeToNow = TimeSpan.FromHours(24)
    };

    public async Task<CampaignKpis> GetCampaignKpisAsync(
        Guid tenantId, Guid campaignId, DateRange range, CancellationToken ct)
    {
        // A per-campaign version token lets a single ingestion event invalidate
        // every cached date range at once — Redis has no safe scan-and-delete
        // at 3,200 req/sec, so we orphan stale keys instead of enumerating them.
        var version = await GetVersionAsync(tenantId, campaignId, ct);
        var key = DataKey(tenantId, campaignId, range, version);

        // Try get.
        var cached = await cache.GetAsync(key, ct);
        if (cached is not null)
        {
            var hit = JsonSerializer.Deserialize(cached, KpiJsonContext.Default.CampaignKpis);
            if (hit is not null)
            {
                return hit;
            }
        }

        // Miss -> query Azure SQL exactly once; everyone else rides the cache.
        var kpis = await repository.AggregateAsync(tenantId, campaignId, range, ct);

        // Set.
        var payload = JsonSerializer.SerializeToUtf8Bytes(kpis, KpiJsonContext.Default.CampaignKpis);
        await cache.SetAsync(key, payload, DataEntry, ct);

        logger.LogDebug(
            "KPI cache miss tenant={TenantId} campaign={CampaignId} range={Range} v={Version}",
            tenantId, campaignId, range, version);

        return kpis;
    }

    // Invoked from the Kafka ingestion consumer when new CampaignEvents land for
    // a campaign. Bumping the version instantly makes every cached range for that
    // campaign unreachable; the orphaned entries age out under the 60s TTL.
    public Task InvalidateOnIngestionAsync(
        Guid tenantId, Guid campaignId, CancellationToken ct)
    {
        var next = clock.GetUtcNow().ToUnixTimeMilliseconds().ToString();
        return cache.SetStringAsync(VersionKey(tenantId, campaignId), next, VersionEntry, ct);
    }

    private async Task<long> GetVersionAsync(Guid tenantId, Guid campaignId, CancellationToken ct)
    {
        var raw = await cache.GetStringAsync(VersionKey(tenantId, campaignId), ct);
        if (raw is not null && long.TryParse(raw, out var version))
        {
            return version;
        }

        // No version yet: seed one so keys stay stable until the next ingestion.
        var seed = clock.GetUtcNow().ToUnixTimeMilliseconds();
        await cache.SetStringAsync(VersionKey(tenantId, campaignId), seed.ToString(), VersionEntry, ct);
        return seed;
    }

    private static string VersionKey(Guid tenantId, Guid campaignId) =>
        $"kpiver:{tenantId}:{campaignId}";

    private static string DataKey(Guid tenantId, Guid campaignId, DateRange range, long version) =>
        $"kpi:{tenantId}:{campaignId}:{range}:v{version}";
}

The read path is pure cache-aside — try Redis, and on a miss run the one expensive aggregate over CampaignEvents, serialize it with a source-generated JsonSerializerContext, and populate the key kpi:{tenant}:{campaign}:{dateRange}:v{version}. Correctness comes from the version token rather than the TTL: InvalidateOnIngestionAsync bumps a per-campaign counter from the Kafka consumer, so newly ingested events orphan every stale range in one write instead of forcing a SCAN/DEL sweep at peak traffic.

The honest stuff: caveats and when it's overkill

1. Don't actually build one — use Redis or Memcached. This is a "could you build it" interview exercise. In production you deploy a battle-tested cache, not a homegrown ring.
2. Cache invalidation is a genuinely hard problem — famously one of the two hard things in computing. TTL is the pragmatic default precisely because perfect invalidation is so hard.
3. A hot key expiring can DDoS your own database. The thundering herd is real; if you skip single-flight/jitter, one expiry can take the DB down. Plan for it explicitly.
4. Write-back trades durability for speed. A node crash loses un-flushed writes — never use it for data you can't reconstruct.
5. Don't cache everything. Caching write-heavy or rarely-read data wastes memory and multiplies staleness. Cache the read-heavy, expensive-to-compute, tolerant-of-staleness data.
6. Stale reads are the price of speed. If your use case truly needs read-your-writes consistency, either invalidate synchronously (write-through) or reconsider whether a cache fits.
7. A single node's failure is a mini-outage without replication. One box down = every key it owned becomes a miss = a load spike on the DB. Replicate the shards.

The model to carry forward

A distributed cache is an O(1) LRU cache, sharded by consistent hashing, kept loosely in sync with a database. The single-node part is a data-structures exercise; the distributed part is all about change — making node joins and failures cheap (consistent hashing + vnodes) and making the gap between cache and DB manageable (write policies + invalidation). The whole thing rests on one permission: the cache is allowed to be a little bit wrong for a little while, and that's what buys the speed.

Three habits this problem teaches:

1. Reach for consistent hashing the moment you shard. hash % N is the answer that gets you a follow-up you can't escape.
2. Name the thundering herd. Anticipating the hot-key-expiry stampede shows you've run a cache, not just read about one.
3. Pick a write policy on purpose. "Cache-aside, invalidate on write" is a one-line answer that settles cache-DB consistency cleanly.

Further reading

• How to Crack Any System Design Interview: A Repeatable Framework
• Design a Rate Limiter — Token Bucket, Sliding Window & Distributed Limits
• We Replaced REST with Kafka and Cut Failures by 90%

---

Prepping for a system design round and want the cache follow-ups drilled — or the next "Design X" walked through the same way? I'm always happy to help; reach me at randhir.jassal@gmail.com.