Distributed Key-Value Cache (Redis / Memcached-like)

Functional

• get(key), set(key, value, ttl?), delete(key). Values are opaque byte strings up to ~1 MB.
• Optional: atomic increment, compare-and-set, multi-get (batch), key expiry.
• Add or remove nodes without a full restart.

Non-functional

• Latency: p99 < 1 ms within a datacenter for get/set (memory + one network hop).
• Throughput: millions of ops/s across the cluster; ~100K+ ops/s per node.
• Scalable: adding a node adds capacity roughly linearly.
• Availability over consistency: a stale read is acceptable; an unavailable cache is not (it must not take the backing store down with it).
• Predictable memory: never exceed a configured limit per node; evict instead of failing.

Out of scope

• Durability (it is a cache, not a database), multi-region replication, complex data types (lists, sets) beyond a mention.

Assumptions

• 1M ops/s cluster-wide, 90% reads. Average key 50 B, average value 1 KB.
• Working set: 500M keys.

Memory

• 500M × (1 KB value + 50 B key + ~50 B overhead) ≈ 500M × 1.1 KB ≈ 550 GB.
• With one replica per shard: 1.1 TB. At 64 GB usable per node (leave headroom for fragmentation and the OS), ~18 shards → ~36 nodes with replicas. Round to 40.

Throughput per node

• 1M ops/s / 18 primaries ≈ 55K ops/s per primary. A single-threaded Redis-style node handles 100K+ simple ops/s, so there is 2× headroom; reads can additionally go to replicas.

Network

• Reads: 900K/s × ~1 KB ≈ 900 MB/s out of the cluster ≈ 7 Gbps, spread over 40 nodes ≈ 180 Mbps each. Fine on 10/25 GbE.
• Writes: 100K/s × 1 KB ≈ 100 MB/s, doubled for replication.

Hot keys

• A key read 100K times/s saturates one node on its own. Plan for hot-key mitigation (client-side caching, key replication) from the start.

Rebalancing

• Adding one node to 18 shards moves ~1/19 ≈ 5% of keys ≈ 29 GB. At 1 Gbps that is a few minutes; cache misses during the move are absorbed by the backing store.

Client library API

• get(key) → value | null
• set(key, value, ttlSeconds?) → ok
• delete(key) → ok
• mget(keys[]) → values[] (grouped by node, sent in parallel)
• incr(key, delta) → newValue (atomic on the owning node)
• cas(key, expectedVersion, value) → ok | conflict

Wire protocol

• Simple binary or RESP-style text protocol over TCP with persistent connections and pipelining. No HTTP overhead on a sub-millisecond path.
• Responses carry a version/CAS token for optimistic concurrency.

Cluster management API

• GET /cluster/ring → current ring: nodes, virtual node positions, epoch number.
• Admin: add node, remove node, rebalance. Changes bump the epoch.

Errors

• MOVED node → the client's ring is stale; refresh and retry. Also NOT_OWNER during migration.
• Timeouts are short (a few ms) and callers treat a timeout as a miss.

In-memory entry

• key (bytes)
• value (bytes), flags, version (for CAS)
• expires_at (absolute time, 0 = none)
• LRU links: prev/next pointers (or a last-access clock for approximated LRU)

Per-node structures

• Hash table: key → entry pointer. O(1) get/set.
• Eviction structure: doubly linked list ordered by recency (LRU) or frequency counters (LFU).
• Expiry: lazy on access plus a periodic sampler that deletes a few expired keys per tick.

Cluster metadata

• ring: sorted array of (hash position, node id) for virtual nodes
• node: id, address, role (primary/replica), slot range or ring positions, health
• epoch: monotonically increasing version of the ring, so clients can detect staleness

Storage choice

• Everything lives in RAM. Memory is allocated in size classes (slabs) to limit fragmentation. Optional append-only log or snapshot only for warm restart, never on the request path.
• Values over a few hundred KB are either rejected or chunked so one key cannot block the event loop.

Why no central router?

• A router adds a hop and a bottleneck. Pushing the ring to clients keeps the path at one network round trip; the ring changes rarely and is small.

1. Consistent hashing with virtual nodes

• Hash each key to a point on a ring of 2^32 (or 2^64) positions; the owner is the first node position clockwise. Adding a node only takes keys from its clockwise neighbour, so roughly 1/N of keys move instead of nearly all with modulo hashing.
• Physical nodes map to 100–200 virtual nodes each, placed at hash(node_id + i). This smooths load imbalance to within a few percent and lets a bigger machine own more virtual nodes.
• Clients learn the ring from the configuration service at startup and watch for changes; a MOVED response with a higher epoch triggers a refresh. During migration, the old owner serves reads and forwards or rejects writes for keys already moved.

2. O(1) LRU and approximated LRU

• Exact LRU: a hash map from key to list node, plus a doubly linked list. get moves the node to the head (unlink + insert, O(1)); set inserts at the head; evict pops the tail. Cost: two pointers (16 B) per entry.
• Redis approximated LRU: store a 24-bit last-access clock per entry, sample N (default 5) random keys on eviction and evict the one with the oldest clock. Nearly as good a hit rate as exact LRU with no list maintenance on every access, which matters because the event loop is single-threaded.
• LFU variant: a logarithmic counter with decay avoids a one-off scan flushing the cache.

3. Replication and failover

• Primary → replica async stream of writes (set/delete). Replica lag is typically milliseconds; a read from a replica may be stale by that much.
• Health checks from the cluster manager (or gossip between nodes). On primary failure: promote the replica, publish a new ring epoch, clients refresh. Writes in the un-replicated window are lost, which is acceptable for a cache.
• Without any replica, a node loss turns 1/N of all reads into misses at once and can overload the backing store. Replicas are a protection for the database, not for the cache's own data.

4. Hot keys

• Detect: per-node top-k key counters (sampled), or client-side sampling reported to a metrics pipeline.
• Mitigate: (a) client-side in-process cache with a short TTL (hundreds of ms) for keys flagged hot; (b) key replication: write the hot key to k nodes under suffixed names key#0..key#k-1 and read a random replica; (c) request coalescing / single-flight: when many callers miss the same key at once, one of them fetches from the backing store and the rest wait on that in-flight result.

5. Memory management

• Slab allocation: memory split into size classes; each entry goes into the smallest class that fits. Avoids fragmentation from mixed value sizes; wastes some bytes per entry.
• Eviction vs TTL: TTL expiry is lazy (checked on read) plus active sampling to reclaim memory. Eviction runs when used memory exceeds the limit, independent of TTL. Both are needed.
• Large values: cap value size (e.g. 1 MB) and chunk above it, otherwise one key monopolises the event loop and the network buffers.

6. Thundering herd on node loss

• When a shard disappears, every cached key on it misses simultaneously and the backing store sees a burst.
• Mitigations: replicas (above); request coalescing so each key is fetched once per process; jittered TTLs so keys never expire together; a circuit breaker in the application that serves stale or degraded responses while the database recovers; warm the replacement node gradually.

Summary of decisions

• Consistent hashing with ~150 virtual nodes per physical node; ring owned by a configuration service and cached in clients.
• Single-threaded event loop per core, hash table plus approximated LRU, lazy + sampled TTL expiry.
• Async primary → replica replication, replica promotion on failure.
• Hot-key handling via client cache, key replication and single-flight.

What was given up

• Consistency: replicas can be stale and un-replicated writes are lost on failover.
• Memory efficiency: replicas halve usable capacity; slabs waste some bytes.
• Simplicity in the client: it must hold the ring and handle MOVED.

With more time

• Persistence for warm restarts, multi-region replication, richer data types, tiered storage (SSD for cold values), TLS and auth, per-tenant quotas.