Routing Metadata Staleness and the Cost of a Wrong Guess

A partitioned store has two kinds of state: the data, and the map from key to the machine holding that data. The second is small, read on every request, and changes whenever a partition splits, merges, or moves. Serving it from a central coordinator on every request would make the coordinator the bottleneck and the single point of latency; caching it in each client means each client eventually acts on a map that is no longer true. The design problem is not how to keep the cache fresh, which is impossible, but how to make a stale entry produce a retry instead of a wrong answer.

The three places the map can live

The routing map is held at three levels, and each level has a different staleness profile.

A coordinator holds the authoritative map and serializes changes to it through a consensus protocol. It is the only place where the map is definitely correct, and it is consulted on cache misses and after rejections. Its cost is a network round trip, so it cannot be in the path of every request.

An intermediate tier, whether proxy nodes, a router service, or the storage nodes themselves, holds a copy of the map and answers client requests by forwarding them. The copy is refreshed by watching the coordinator or by polling. Its staleness is bounded by the refresh interval, typically a fraction of a second for a watch-based scheme and seconds for polling.

The client holds a copy, refreshed on a timer or on demand after a redirect. Client-side caching is what makes a routing tier unnecessary for simple deployments, and it is what makes staleness hardest to bound, because the client’s refresh interval is a property of the client’s configuration and its network reachability, not of the cluster’s policy.

A request that uses a stale map fails in one of four ways, and only one of them is acceptable.

It reaches the right node, because the partition did not move. No harm. It reaches a node that still holds the data but no longer owns it. The node can answer with a redirect to the current owner, which costs one extra round trip. It reaches a node that has already dropped the data. The node cannot answer and must respond with an error that the client interprets as “refresh and retry”. It reaches a node that holds a copy of the data from before a write that happened elsewhere. This is the dangerous case, because the node can answer confidently and wrongly.

The first three are latency problems. The fourth is a correctness problem, and preventing it is the reason ownership changes carry epochs.

Epochs make ownership verifiable

An epoch is a counter attached to a partition’s ownership record and incremented on every ownership change. The record is (partition, owner, epoch), and the coordinator’s copy is authoritative.

Every request carries the epoch the client believes is current. A node compares the request’s epoch against the epoch it holds for that partition and rejects anything lower. The comparison is a single integer check, so the cost is negligible; the effect is that a node which has lost ownership cannot serve a request from a client that has not noticed.

import time

class StaleEpoch(Exception):
    def __init__(self, current):
        self.current = current          # the client can cache this and retry

class LeaseExpired(Exception):
    pass

class Owner:
    def __init__(self, ownership, lease_deadline):
        self.ownership = ownership            # partition -> (owner, epoch)
        self.lease_deadline = lease_deadline  # partition -> monotonic deadline

    def admit(self, partition: int, epoch: int) -> None:
        held = self.ownership.get(partition)
        if held is None or epoch < held[1]:
            raise StaleEpoch(held[1] if held else None)
        if time.monotonic() > self.lease_deadline[partition]:
            raise LeaseExpired()    # still holds the data, must not serve it

The rejection carries the current epoch, so the client can update its map from the response without another lookup. This is the “redirect” mechanism in its cheapest form: the error itself is the fresh routing information, and the client’s retry goes directly to the new owner. A client that receives a rejection with a higher epoch it did not know about can cache that and proceed, which means the map converges through use rather than through polling.

The epoch must be monotonic per partition, not global. A global counter would force every ownership change to serialize through one sequence, and it would grow without bound across the cluster’s lifetime. A per-partition counter is enough because comparisons only ever happen within one partition’s ownership history.

The epoch also gives the coordinator a cheap way to fence a node that has been partitioned from the network. A node that cannot reach the coordinator cannot learn that its epoch has been superseded, so it continues to believe it owns the range. If it can still reach clients, it will serve reads from its copy. The defence is a lease: the node’s ownership is valid only until a deadline it must keep renewing with the coordinator. Once the deadline passes without renewal, the node stops serving, even though it has not been told to. This is the standard lease construction, and it is the same mechanism that makes a lock service safe against a paused client. The admit method above checks both conditions in one place, because a node that has lost its lease and a node that has lost its epoch are both unsafe in the same way.

A node that has stopped serving but still holds the data is in a safe state: it cannot answer wrongly, and it cannot accept writes that the new owner will not see. When it regains contact with the coordinator it learns the new epoch and can rejoin as a replica.

The window between change and convergence

The staleness that matters is the time from the coordinator committing an ownership change to the last client that would have used the old owner having updated. That window is not bounded by the coordinator; it is bounded by the slowest client’s refresh behaviour, which may be a client that is idle, or one that is on a slow link, or one that is not running at all.

A client that is not running is harmless, because it issues no requests. A client that is running but has not refreshed will issue requests with a stale epoch and receive rejections. The rejections cost a round trip each, and they cost the new owner nothing, because they are rejected by the old owner before reaching the storage layer.

The failure mode to design against is a client that caches the map with no expiry and refreshes only on error. Such a client is correct, because the first stale request after a change produces an error and a refresh. But it is correct only if the old owner rejects rather than answers, and that is exactly what the epoch check provides. A system without epochs, where the old owner answers with whatever it has, converts the client’s cached map into a source of stale reads with no bound on their staleness.

The interval also matters for write retries. A write that is rejected for a stale epoch has not been applied, so retrying it against the new owner is safe. A write that is accepted by the old owner during the interval before the ownership change commits has been applied, and the migration protocol must ensure it reaches the new owner. This is the dual-write requirement described in Shard Migration Cutover and the Dual-Write Window, and the epoch is what lets the client distinguish “rejected, safe to retry” from “accepted, already durable”.

Redirects versus lookups

Two client strategies exist and they have different cost profiles.

Lookup first. The client asks the coordinator for the owner of the key, then sends the request. Every request costs two round trips unless the client caches. With a cache and a long-lived mapping, most requests cost one round trip and the cache miss costs two. The advantage is that the client never sends a request to a node that cannot serve it, so the coordinator sees a bounded load proportional to cache misses.

Send and redirect. The client sends to a node it believes owns the key. If wrong, the node answers with a redirect and the client retries. The first request after a change costs two round trips, and the steady state costs one. The advantage is that the coordinator is out of the data path entirely; the disadvantage is that redirects consume capacity on nodes that are already busy, and a widespread map change produces a burst of redirects concentrated on the nodes that just lost data.

The redirect strategy degrades badly during a rebalance, because every client with a stale map redirects at once, and the nodes losing partitions are the ones fielding the redirects while also streaming the data out. The lookup strategy degrades more gracefully, because the coordinator absorbs the burst, and coordinators are usually replicated and provisioned for metadata traffic rather than for data.

A hybrid is common: the client caches, and on rejection it re-resolves through the coordinator rather than trusting the epoch in the error. That costs an extra round trip on every rejection but guarantees the client’s map is complete rather than patched, which matters when several partitions moved at once. A client that patches its map from error responses one partition at a time may need several rejections to converge after a multi-partition move.

The metadata service as a dependency

Once routing lives in a consensus-backed service, every request path depends on that service being reachable, at least on cache misses. A client that cannot reach the metadata service and has a cached map can continue, using the map it has, until an epoch changes. A client that cannot reach it and has no map cannot start.

This makes the metadata service’s availability a prerequisite for the cluster’s availability, and it is why the service is usually replicated across failure domains with a small quorum, and why clients cache aggressively. The alternative, embedding the map in the data nodes and gossiping it, removes the central dependency but replaces it with a convergence problem: two nodes may disagree about who owns a range, and there is no arbiter to settle the disagreement. The gossip design is workable when ownership changes are rare and when a disagreement resolves to “both refuse”, but it cannot support a fast rebalancer that moves many partitions concurrently.

The size of the map also constrains the design. A cluster with a hundred thousand partitions has a map of a hundred thousand entries, each with a key range, an owner, and an epoch. At roughly a hundred bytes per entry that is ten megabytes, which is large for a per-client cache but not impossible. At a million partitions it is a hundred megabytes, and per-client caching stops being viable; clients must then resolve through a routing tier that holds the map once, which reintroduces the intermediate hop. This is a concrete reason to keep partitions coarse, as argued in Range Partition Boundaries and the Split Key Contract.

Failures that look like data loss

Three symptoms are commonly misdiagnosed as corruption and are in fact routing problems.

A read returns nothing for a key that exists. The client’s map points at a node that has already dropped the range, and the node answered “not found” instead of rejecting the stale epoch. A node that has dropped a range must remember that it once owned it, at least for a grace period, so it can distinguish “I never had this key” from “I no longer serve this range”. Without that memory, the node cannot tell the client to refresh.

A read returns an old value. The client’s map points at a node that still holds the data as a replica and answers reads without checking ownership. Replicas must reject reads that arrive with an epoch they have been superseded on, or the store’s read path must route through the owner.

A write is acknowledged and then vanishes. The write reached a node that accepted it under an epoch that was about to be superseded, and the migration did not carry it forward. This is the only one of the three that is genuinely a loss, and it is prevented by the dual-write rule rather than by the epoch.

The distinction matters operationally because the first two are fixed by a refresh and the third is not. A system that reports all three identically trains its operators to ignore the signal, which is how a real loss goes unnoticed.