Distributed Cache

A distributed, sharded LRU cache built in Go — implemented bottom-up to understand the concurrency and distribution problems a real cache has to solve, and to measure (rather than assume) where the bottlenecks actually are.

This is a learning-driven build: every component is written from first principles, tested, and — where performance is claimed — benchmarked with repeated measurements rather than single samples.

Objective

Build a horizontally-scalable cache that a client can hit over HTTP, where:

• each node holds a fast, correct, concurrency-safe in-memory LRU cache, and
• a cluster of nodes partitions the keyspace via consistent hashing, so capacity scales with the number of nodes and adding/removing a node remaps only a small fraction of keys.

The project deliberately separates two distinct sharding problems that are often conflated:

1. Lock sharding inside a node (Phase 1) — splitting one node's keyspace across independent locks to remove mutex contention. Uses modulo hashing, because the shard count is fixed at construction.
2. Key sharding across nodes (Phase 2) — routing keys to the node that owns them. Uses consistent hashing with virtual nodes, because node membership changes and we want minimal remapping when it does.

Same partitioning idea, two different algorithms, each chosen by whether the bucket set is static or dynamic.

Architecture

Client
  │
  ▼
Load Balancer            (round-robin: app servers are stateless)
  │
  ▼
Router                   (consistent hashing: picks the node that owns the key)  ← Phase 2
  │
  ├──────────┬──────────┐
  ▼          ▼          ▼
Node 1     Node 2     Node 3         (HTTP server per node)  ← Phase 1
  │
  ▼
ShardedCache             (N shards, one mutex each — modulo hashing)  ← Phase 1
  │
  ▼
LRU Cache                (hashmap + sentinel doubly-linked list, O(1))  ← Phase 1

Dependency direction points inward — the LRU core depends on nothing; the HTTP layer depends on the cache; nothing depends back on the delivery mechanism. Each package has a single responsibility.

cmd/node      → internal/node → internal/cache → internal/lru
cmd/router    → internal/ring                                   (Phase 2)

Status

Phase 1 — single node: complete.

• internal/lru — O(1) LRU: hashmap for lookup, doubly-linked list with sentinel nodes for ordering/eviction, the map storing pointers into the list so location hands off directly to O(1) reordering. Behavior proven by tests (hit, miss, update, eviction, recency).
• internal/cache — two interchangeable concurrency strategies over the LRU:
  • SafeCache: single sync.Mutex (baseline).
  • ShardedCache: N shards, each an independent LRU + mutex, key routed by FNV-1a hash modulo shard count.
• internal/node, cmd/node — HTTP server exposing GET /get?key= and POST /set, with typed request/response contracts and method-enforced routing.

Phase 2 — multi-node cluster: not started.

Benchmarks & Findings

The central question of Phase 1: does the cache's lock become a bottleneck under concurrency, and does sharding help? The answer turned out to depend entirely on where the measurement is taken.

1. Under HTTP load, the lock is invisible

Load-tested the single-mutex node with wrk (three workloads, randomized keys):

Workload	Req/sec	p50	p99
Read-heavy	112,624	381µs	0.87ms
Write-heavy	111,810	382µs	0.93ms
Mixed 90/10	112,252	384µs	0.87ms

Write-heavy — which takes the exclusive lock on every request — came within 0.7% of read-heavy. Raising connections 10× (50 → 500) produced zero extra throughput (~111k req/s flat) while latency rose linearly. That is the signature of a saturated bottleneck that is not the lock: by Little's Law, with throughput pinned, latency is proportional to in-flight requests. The bottleneck is the per-request HTTP/syscall/TCP path, whose cost (microseconds) dwarfs the nanosecond-scale critical section.

Conclusion: HTTP overhead masks the lock. To measure lock contention, HTTP must be removed.

2. In-process, the lock is a real bottleneck — and sharding fixes it

Benchmarked both caches in-process (testing.B + RunParallel, no HTTP), 90/10 read/write, -count=5 -benchtime=3s, GOMAXPROCS 1→8 (Apple M4):

Cores	SafeCache (mutex)	ShardedCache (16)	Sharded vs mutex
1	23.7 ns/op	26.8 ns/op	13% slower
2	51.2 ns/op	31.9 ns/op	38% faster
4	91.7 ns/op	29.3 ns/op	68% faster
8	100.3 ns/op	43.3 ns/op	57% faster

The single mutex degrades 4.2× from 1 to 8 cores — goroutines serialize on one lock. Sharding stays nearly flat and is 2.3–3.1× faster under contention.

Two details worth noting:

• Sharding is slower with no contention (cpu=1). The FNV hash to locate a shard is ~3ns of pure overhead when there is no lock to contend for. Sharding is a trade: a small constant per-op cost bought in exchange for parallelism. It is a pessimization for single-threaded workloads.
• ShardedCache's curve rises slightly at 8 cores (43ns). With 8 goroutines over 16 shards, occasional shard collisions still cause contention (birthday-paradox). Increasing shard count trades memory for further contention reduction.

The takeaway

The bottleneck depends on what else is in the request path. The same lock is invisible behind HTTP and dominant in-process. Optimizing the lock only matters once the network is not the constraint — the correct engineering move was to measure both layers rather than assume the mutex was the problem.

Methodology note: a single benchmark run is not a result. Early single-shot runs of the identical code disagreed by 2–3×; only repeated runs (-count) with quiet-machine conditions produced trustworthy numbers.

Roadmap

Phase 2 — Multi-node cluster (next)

Turns this from a fast single node into an actual distributed cache: keys are partitioned across nodes, routed by consistent hashing.

• internal/ring — consistent hash ring with virtual nodes; AddNode, RemoveNode, GetNode(key), plus a distribution test proving keys spread evenly (and empirically demonstrating why first-letter/prefix sharding would create hot shards).
• cmd/router — routes each request to the owning node via the ring; several node processes run behind one router. Membership is static at this stage (the router is configured with the node set at startup).
• Distributed load test — verify even key distribution across the cluster.

Exit criteria: multiple node processes, a router that consistent-hashes keys to them, and a load test showing even distribution.

Phase 3 — Fault tolerance and self-healing

This phase makes the cluster survive node failure. It commits to a leaderless, AP-oriented design (in the Dynamo/Cassandra lineage): no central coordinator, membership managed by gossip, availability favored under partition.

The committed arc, built and measured in this order:

1. Replication — each key lives on N nodes (the next N−1 nodes clockwise on the ring), so a single node's death does not lose data.
2. Gossip-based membership + failure detection — nodes self-discover and detect deaths by periodically exchanging membership state with random peers; no coordinator. Failure detection via heartbeat timeout (with phi-accrual as a possible refinement).
3. Failure demonstration (the centerpiece) — kill a node under load and measure: rounds/latency for the cluster to converge on the death via gossip, and read availability during the failure (reads still succeed because replicas exist). This is Phase 3's equivalent of Phase 1's contention benchmark — a measured experiment, not just a feature.

Supporting items as they fit the arc:

• Rebalancing — when membership changes, key data actually moves to its new owner (the ring says who should own a key; rebalancing ships the data). Pairs with gossip.
• WAL / persistence — survive process restart, not just node loss: write-ahead log plus periodic snapshot, rewarm on boot. Self-contained.

The road not taken (deliberately): the alternative to this leaderless design is a CP, coordinator-managed approach — master/slave per shard with leader election and fencing tokens on master failure, membership held in a strongly-consistent store (ZooKeeper/etcd). That is a coherent design with opposite trade-offs (consistency over availability under partition). This project chooses the AP/gossip path; the CP path is noted as the principled alternative rather than a second thing to build, because the two are competing architectures, not complementary features.

Running it

# start a node
go run ./cmd/node        # listens on :8080

# set and get
curl -X POST localhost:8080/set -d '{"key":"a","value":"apple"}'
curl "localhost:8080/get?key=a"

# tests and benchmarks
go test ./...
go test -bench=. -cpu=1,2,4,8 -benchtime=3s -count=5 ./internal/cache/