This post follows part 2 of the series, in which I explored cache eviction strategies, and is followed by part 4 of the series.

The 'What' and 'Why' of sharding

Currently, my cache runs on one machine. It's a key-value cache so far, but it won't be until our next step that I can truly live up to the name of the series by having a distributed key-value cache. That's because it's now time to implement distributed sharding.

Sharding is where we distribute our cache across multiple processes/machines, each of which we call a 'shard' since they each only hold a portion of the total data. The redundancy is reminescent of the horcruxes from Harry Potter, but sharding is more like if Voldemort split up his memories evenly among his horcruxes.

What does this do? Many things!

Firstly, increased potential capacity: if our cache needs to store titanic amounts of data, sharding becomes essential. Even if we get a bigger and more expensive machine (vertical scaling) to host our cache, it has its limits at massive scale. Consider, for example, if every tweet ever written was crammed into one machine. As magical as {production machines](/blog/prod-server.md) are, they are not like a bag of holding!

So intead, we store more data by grouping together machines (horizontal scaling), and shard our data among all of them. Yeah, my little cache isn't going to get millions or billions of users, but at least it'll be prepared for it!

Similarly, another benefit is higher throughput. If requests are spread across multiple nodes, then there's less competition for CPU, network bandwidth, locks, and memory.

There're more benefits I can mention (one of which might slightly spoil the subsequent blog post's hero!), but let's move on and start figuring out how to shard.

Consistent Hashing

We need one answer that every cache node and every client can independently calculate: given a key, which node owns it? The simplest possibility is hash(key) % numberOfNodes, but that has a nasty scaling problem. If I change from three nodes to four, the divisor changes, so a large portion of keys land on different machines.

Consistent hashing takes a more interesting approach. It treats the hash space as a circle, or ring. I place both cache nodes and cache keys somewhere on that ring, then give each key to the first node encountered while moving clockwise. Adding a node only changes ownership for the slice of the ring immediately before that new node, instead of reshuffling the entire cache.

A consistent-hashing ring maps keys to shards

Hash functions return numbers in a fixed range. The circle is just a useful way to visualize that range: after the largest hash value, we wrap around to zero again.

Building the ring

I start by describing a cluster node with an ID and its base HTTP URL. The ID is what goes on the ring; the URL is where the client sends the request once it knows the owner.

type Node struct {
	ID      string
	BaseURL string
}

type ringPoint struct {
	hash uint64
	node Node
}

type HashRing struct {
	points []ringPoint
	nodes  map[string]Node
}

The implementation uses SHA-256 (a common hash function), then reads the first eight bytes as one unsigned 64-bit number. That gives every string, either a cache key or a node identifier, a deterministic position in the same $2^{64}$-sized hash space.

func hashString(value string) uint64 {
	sum := sha256.Sum256([]byte(value))
	return binary.BigEndian.Uint64(sum[:8])
}

One point per physical node is not enough for a well-balanced ring. With a small cluster, random placement can leave one node responsible for a huge arc while another owns barely anything, just due to chance. To smooth that out, each cache node gets 128 virtual nodes. They are not extra processes; they are simply 128 independent positions on the ring that all point back to the same physical node. Like stand-in representatives for the physical machine they correspond to.

const defaultVirtualNodes = 128

for replica := 0; replica < virtualNodes; replica++ {
	points = append(points, ringPoint{
		hash: hashString(node.ID + "#" + strconv.Itoa(replica)),
		node: node,
	})
}

sort.Slice(points, func(i, j int) bool {
	return points[i].hash < points[j].hash
})

The node.ID + "#" + replica string makes every virtual-node position stable across restarts. Every node and every client must build the ring from the exact same membership list, so they agree on ownership without having coordinate for every request!

Finding a key's owner

Once the ring points are sorted, looking up an owner is best done with binary search. First hash the key, then find the first ring point whose hash is greater than or equal to the key hash. That is our clockwise walk. If the key sits past the final point, the ring wraps around and the first point owns it.

func (r *HashRing) Owner(key string) Node {
	keyHash := hashString(key)
	index := sort.Search(len(r.points), func(i int) bool {
		return r.points[i].hash >= keyHash
	})
	if index == len(r.points) {
		index = 0
	}
	return r.points[index].node
}

That means routing is $O(\log(n \times v))$, where $n$ is the number of physical nodes and $v$ is the number of virtual nodes assigned to each physical node. In my project, the ring is built at startup and stays fixed, so lookups only read the sorted slice.

Routing requests to the shard

To run locally, I configure each node through environment variables. CACHE_NODES lists the complete topology in id=url form, CACHE_NODE_ID identifies the process currently running, and CACHE_VIRTUAL_NODES can override the default of 128.

CACHE_NODES=node-01=http://127.0.0.1:8081,node-02=http://127.0.0.1:8082
CACHE_NODE_ID=node-01
CACHE_ADDR=:8081
CACHE_CAPACITY=100

Applications do not pick a server themselves. ShardedClient asks the ring for the owner, URL-escapes the key, and sends the CRUD request directly to that node:

owner := c.ring.Owner(key)
requestURL := owner.BaseURL + "/kv/" + url.PathEscape(key)
request, err := http.NewRequestWithContext(ctx, method, requestURL, body)

Each cache server also calculates ownership before it handles /kv/{key}. If a request reaches the wrong node directly, it responds with 421 Misdirected Request instead of accidentally storing a duplicate copy of the key on the wrong shard.

if ring != nil && ring.Owner(key).ID != nodeID {
	http.Error(w, "key belongs to another cache node", http.StatusMisdirectedRequest)
	return
}

To handle unexpected failures and mismatches, the parser rejects missing IDs, duplicate IDs, duplicate URLs, and malformed HTTP URLs early. There's really a lot to consider when scaling out to a distributed system! So much more capability, but it's easy to lose control without careful design.

Cold misses

One limitation of my current implementation so far: my cluster does not migrate data when membership changes yet. Imagine that user:42 is currently cached on node A, then I add node D. After every client rebuilds its ring, consistent hashing may decide that node D is now the owner of user:42. The value is still physically sitting in node A's memory, but nobody copies it to node D, and clients no longer ask node A for it. They route the next read to node D instead. But the data isn't there. This is called a cold miss. In production, for a cache that's paired with a source of truth, like a database, this can mean querying the database after the cache miss, then repopulating the cache with the result from the database. Since this 'cache-aside with cold miss' strategy is valid design, I'm alright with this limitation.

Testing

With the exponentially increasing complexity of this project, it was about time I started writing some tests. Having worked with distributed systems before, I am fully aware of how, no matter what, writing tests translates to LESS development time, not more. If you know, you know.

Some invariants I validate: The same key must always map to the same owner; keys should be distributed across the available nodes; and, most importantly, adding a node should move fewer keys than the naive modulo approach.

before := mustNewHashRing(t, testNodes(3))
after := mustNewHashRing(t, testNodes(4))

for i := 0; i < 10_000; i++ {
	key := fmt.Sprintf("key-%d", i)
	if before.Owner(key).ID != after.Owner(key).ID {
		consistentMoves++
	}
	if hashString(key)%3 != hashString(key)%4 {
		moduloMoves++
	}
}

In the end...

When a shard node is down, we return a 503 error: "Service Unavailable". But what if one node going down didn't have to mean its data was lost? In the next post in this series, I'll explore the key oppurtunities that our horizontal scaling has enabled: reliability via replication!

Building a Distributed Key-Value Cache part 3/6: Sharding