This post follows part 3 of the series, in which I introduce horizontal scaling through cache sharding, and is followed by part 5 of the series.
My cache is finally distributed, but there is still a rather large hole: when a node dies, the system breaks. In the previous post, I used consistent hashing to give each key an owner. This time, I’ll keep that ownership model, but give each key several possible homes. If one node fails, another replica can answer instead, keeping the system alive until the node can be brought back or replaced with a clone.
In other words, we want our system to cheat death with necromancy! Let's learn how this dark art can be performed.
Step 1: Choosing the replicas
The hash ring already tells us where a key belongs. I can extend that same clockwise walk to find multiple distinct nodes instead of stopping at the first one.
func (r *HashRing) Replicas(key string, replicationFactor int) []Node {
if replicationFactor <= 0 {
return nil
}
if replicationFactor > len(r.nodes) {
replicationFactor = len(r.nodes)
}
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
}
replicas := make([]Node, 0, replicationFactor)
seen := make(map[string]struct{}, replicationFactor)
for offset := 0; len(replicas) < replicationFactor; offset++ {
node := r.points[(index+offset)%len(r.points)].node
if _, exists := seen[node.ID]; exists {
continue
}
seen[node.ID] = struct{}{}
replicas = append(replicas, node)
}
return replicas
}The first distinct node is the leader for the key. The next distinct clockwise nodes are its replicas. The word “distinct” matters because every physical node has 128 virtual nodes on the ring. Without deduplication, one physical machine could appear several times in a row and use up the entire replication factor by itself.
I configure the replication factor with CACHE_REPLICATION_FACTOR. A factor of 1 behaves like the old single-owner setup. A factor of 3 stores each key on three different cache nodes, assuming the cluster has at least three nodes.
Step 2: Letting nodes accept replica traffic
Previously, a node rejected any request for a key that it did not own. That was useful for preventing accidental duplicate writes, but now it is too strict: a replica also needs to accept requests for the keys it stores.
if ring != nil && !ring.IsReplica(key, nodeID, factor) {
http.Error(w, "key belongs to another cache node", http.StatusMisdirectedRequest)
return
}The node still rejects unrelated keys. It just now recognizes the full replica set as valid. Every cache node and client must use the same CACHE_NODES configuration, or they may construct different rings and disagree about who is allowed to store a key. Distributed systems are apparently very sensitive to everyone having the same map.
Step 3: Writing to a quorum
Writing to one replica would not be enough. If the leader accepts a write and immediately disappears before the other replicas receive it, a failover read could return stale data or a miss.
Instead, a replicated write sends the mutation to every configured replica and waits for a majority quorum:
func (c *ReplicatedClient) write(
ctx context.Context,
method, key, value string,
expected int,
) error {
quorum := c.replicationFactor/2 + 1
successes := 0
for _, node := range c.Replicas(key) {
response, err := doNodeRequest(ctx, c.httpClient, node, method, key, value)
if err != nil {
continue
}
if response.StatusCode != expected {
return newStatusError(response)
}
response.Body.Close()
successes++
}
if successes >= quorum {
return nil
}
return ErrOwnerUnavailable
}For a replication factor of 3, at least 2 replicas must acknowledge the mutation. This lets one replica be unavailable while the write still succeeds. With a factor of 2, the quorum is also 2, so both replicas need to be reachable. The odd number is more useful here because it gives us a majority without requiring every node.
There is a subtle tradeoff. A successful quorum write does not guarantee that every replica has the newest value. One node may have been offline when the write happened. I have durability of a sort, but not yet a mechanism to repair stale replicas when they come back. I'll leave it out of scope to implement, but one solution would be read repair: when a client queries for a key and gets back responses from more than one node, each should send a version number too, and out of date responses are sent back the most up-to-date response by the client. Another alternative solution is an anti-entropy process, a background process that looks for differences in data in replicas to perform updates.
Step 4: Failing over reads
Reads walk the same replica list, trying the leader first and then the remaining replicas:
func (c *ReplicatedClient) Read(ctx context.Context, key string) (string, error) {
for _, node := range c.Replicas(key) {
response, err := doNodeRequest(ctx, c.httpClient, node, http.MethodGet, key, "")
if err != nil {
continue
}
if response.StatusCode == http.StatusNotFound {
response.Body.Close()
continue
}
if response.StatusCode != http.StatusOK {
return "", newStatusError(response)
}
var value valueRequest
if err := json.NewDecoder(response.Body).Decode(&value); err != nil {
response.Body.Close()
return "", err
}
response.Body.Close()
return value.Value, nil
}
return "", &StatusError{
StatusCode: http.StatusNotFound,
Message: "key not found",
}
}If the leader is unavailable, the next reachable replica effectively becomes the leader for that request. Nothing permanently changes in the ring; this is request-level failover rather than a membership protocol that rewrites the cluster. The cache can keep serving reads even when one of its preferred nodes has vanished. It is a little like necromancy, one of my favorite types of magic: we are not bringing the dead node back, but we are raising a surviving copy of its data to answer in its place.
Step 5: The replicated client
The existing ShardedClient still has one job: route each request to one owner. I added a separate ReplicatedClient so callers can opt into replica behavior without changing the semantics of the original client.
type ReplicatedClient struct {
ring *HashRing
replicationFactor int
httpClient *http.Client
}
func (c *ReplicatedClient) Leader(key string) Node {
return c.ring.Replicas(key, c.replicationFactor)[0]
}
func (c *ReplicatedClient) Replicas(key string) []Node {
return c.ring.Replicas(key, c.replicationFactor)
}This separation keeps the system flexible. A caller that wants the simplest possible route can use ShardedClient. A caller that wants failover can use ReplicatedClient. Both clients still speak the same /kv/{key} HTTP API, so the cache nodes do not need a second set of CRUD handlers.
Step 6: Testing failure
I added tests for the new replica behavior:
- replicas should be distinct physical nodes
- the leader should be the first clockwise replica
- replication factors larger than the cluster should be rejected
- writes should succeed once a majority acknowledges them
- writes should fail when a quorum is unavailable
- reads should fall through to another replica when the leader is unavailable
- a node should accept a key when it belongs to the configured replica set
Cold misses are still cold
Replication solves a different problem from membership changes. If I add a new node and rebuild the hash ring, some keys may have a new replica set. The old nodes still hold their previous in-memory values, and the new node starts empty. A request routed to the new node can still be a cold miss.
Similarly, if a replica was offline during a successful quorum write, it may return an older value when it comes back. The next thing I need is a repair or synchronization process that can compare replicas and repopulate stale ones.
What has our necromancy-like replication handling accomplished?
The cache now has a better answer to the question, “What happens when a node dies?” Instead of immediately returning 503 Service Unavailable, the client can try another replica. Writes require a majority, reads can fail over, and each key has a deterministic leader plus a deterministic set of backup nodes.
It is not yet a fully self-healing system. There is no automatic data migration, membership service, or background replica repair. But the cache has crossed an important line: losing one machine no longer automatically means losing access to every key that machine owned.
Next, I’ll make the data survive something even more dramatic than a dead node: a process restart. Read about part 5 of the series next!