Replication, Sharding & Consistent Hashing · How it works
11 min readHow it works
Replication scales reads, not writes
A primary accepts writes and streams them to replicas that serve reads. This helps a read-heavy system enormously and provides a failover target, but the write path is unchanged: one machine still applies every write.
The choice between synchronous and asynchronous replication is the durability-versus-latency trade, and it is worth stating in terms of the two recovery objectives.
| Write latency | Includes the replica round trip | Primary only |
|---|---|---|
| Data loss on primary failure (RPO) | Zero | Up to the replication lag |
| Availability risk | A slow replica slows or blocks writes | None |
Semi-synchronous replication is the common compromise: acknowledge once at least one replica has the write, which bounds loss to that one replica's failure while not waiting for all of them.
A: Because every write must still be applied by the single primary, and replicas are downstream consumers of that same write stream rather than additional write capacity. In fact adding replicas makes the write path slightly worse: the primary has to ship the log to more destinations, consuming network and CPU it would otherwise spend on writes, and with synchronous replication it must also wait for acknowledgements. So a system saturated on writes gets no relief and marginally more load. The mechanisms that do address write throughput are reducing the write volume itself — batching, coalescing repeated updates, moving high-frequency low-value writes such as counters into a cache or a stream and aggregating before persisting — or partitioning the data so that different writes are handled by different primaries, which is sharding. It is worth being precise about this in an interview because "add replicas" is a common reflex answer to any scaling question, and identifying that the constrained resource is write capacity rather than read capacity is the actual insight.
Replication lag and the read-your-own-writes problem
Asynchronous replication means a replica is always slightly behind. Usually that is fine. It stops being fine the moment a user reads data they just wrote: they update their profile, the read goes to a lagging replica, and they see the old value. This reads as a lost update, and users respond by doing it again.
The standard fixes, roughly in order of cost:
Route a user's reads to the primary for a short window after they write — simple, and it concentrates a little extra load on the primary. Track the write position (a log sequence number) in the user's session and only use a replica that has caught up past it — more precise, and it requires the replica to expose its position. Or read from the primary only for the specific views that follow a write, which is the surgical option and requires knowing your access patterns.
A: The write went to the primary and the subsequent read was served by an asynchronous replica that had not yet applied it, so the user is shown a version of the world from a moment before their own change — the read-your-own-writes anomaly. It is particularly damaging because it looks exactly like a failed save, so users retry, which doubles the write load and can create duplicate records if the operation is not idempotent. The cheapest fix is to route reads to the primary for a bounded window, typically a few seconds, after that user performs a write, using a session flag or a cookie; it is simple and correct, at the cost of some extra primary load. A more precise approach is to capture the replication position at the time of the write, store it in the session, and require any replica serving that user to have applied at least that position, falling back to the primary if none has — this gives correctness without sending all post-write reads to the primary. A third option, which is often the best user experience, is to update the client optimistically from the response of the write itself rather than re-fetching, so no read is needed at all. I would also alert on lag, because a lag large enough for users to notice usually indicates a long-running transaction or write saturation rather than normal operation.
Sharding, and the decision you cannot undo
Sharding splits data across independent databases, each owning a subset. Writes for different shards proceed in parallel, which is what makes write throughput scale.
What you give up is significant and should be stated plainly: a join across shards has to be done in the application, a transaction across shards needs a distributed protocol or a saga, and any query that does not include the shard key must be fanned out to every shard and merged — which is slower than the unsharded version was, and gets slower as you add shards.
The three common strategies:
Range-based partitions by key ranges (A–F, G–M...). Range scans stay efficient, but the distribution is only as even as your data, and sequential keys such as timestamps send all new writes to the last shard — a permanent hot spot.
Hash-based hashes the key to choose a shard. Distribution is even, but range queries are destroyed because adjacent keys land on different shards.
Directory-based keeps an explicit lookup service mapping keys to shards. Maximum flexibility, including per-tenant placement and easy rebalancing, at the cost of a lookup on every operation and a new critical dependency.
A: Because it determines which queries stay efficient and which become fan-outs, and it is extremely expensive to change once data is distributed — altering it means rewriting essentially every row into a new placement, usually with dual-writes and a long backfill while the system stays live. A well-chosen key means the overwhelming majority of queries include it and therefore touch exactly one shard, which is what makes the system fast. A poorly-chosen one means common queries lack it, so each has to be broadcast to every shard and merged in the application, and that pattern gets worse with every shard added — the opposite of scaling. The key also determines distribution: if it has low cardinality or skewed values, some shards receive far more data and traffic than others, and a single-shard bottleneck limits the whole system no matter how many shards exist. And it determines what can be transactional, since atomicity is normally only available within a shard, so entities that must change together should share a key. This is why the honest interview answer starts from the access patterns and the entities that need atomicity, then derives the key, rather than picking something evenly distributed and discovering the query implications later.
Consistent hashing
The naive way to map keys to nodes is hash(key) % N. It distributes evenly and is trivial — and it fails badly the moment N changes. Going from 10 nodes to 11 changes the divisor, so nearly every key maps somewhere new. For a cache that means the hit rate drops to almost zero at once; for a database it means moving nearly all of the data.
Consistent hashing arranges the hash output into a ring. Each node is placed at one or more points on the ring, and each key belongs to the first node encountered clockwise from the key's hash position. Adding or removing a node only affects the keys in the arc between it and its neighbour — on average 1/N of the keyspace, not all of it.
# Consistent hashing with virtual nodes. The ring is a sorted list of
# (position, node) and lookup is a binary search for the first position >= hash.
import bisect, hashlib
class HashRing:
def __init__(self, nodes, vnodes=150):
# vnodes matter: with one point per node, random placement leaves
# some arcs far larger than others, so load is badly uneven.
self._ring = []
self._vnodes = vnodes
for node in nodes:
self.add(node)
def _hash(self, key):
return int(hashlib.md5(key.encode()).hexdigest(), 16)
def add(self, node):
for i in range(self._vnodes):
bisect.insort(self._ring, (self._hash(f"{node}#{i}"), node))
def remove(self, node):
self._ring = [(p, n) for p, n in self._ring if n != node]
def get(self, key):
if not self._ring:
return None
pos = self._hash(key)
idx = bisect.bisect(self._ring, (pos,))
# Wrap around: past the last point, the owner is the first node.
return self._ring[idx % len(self._ring)][1]
The virtual-node count is the detail most people omit. With one ring position per node, the arcs between randomly placed points vary widely, so some nodes own much more of the keyspace than others. Placing each node at 100–200 positions averages the variance out and gives roughly even distribution. Virtual nodes also let you weight heterogeneous hardware by giving a larger machine more positions.
A: It solves the remapping problem in hash(key) % N: because the node count is the divisor, changing it changes the result for nearly every key, so adding or losing a single node reshuffles almost the entire keyspace. For a cache that means an instantaneous collapse of the hit rate and a full-load stampede onto the origin; for a sharded datastore it means moving nearly all the data. Consistent hashing places nodes and keys on a ring and assigns each key to the next node clockwise, so adding or removing a node only reassigns the keys in that node's arc — about 1/N of the keyspace — while every other key stays exactly where it was. Virtual nodes fix the distribution weakness of the basic scheme: with a single point per node, the randomly sized arcs mean some nodes own several times more of the keyspace than others, so load is uneven no matter how many nodes you add. Giving each node many positions — typically one to two hundred — averages the arc sizes so ownership is close to even, spreads a departing node's keys across many remaining nodes rather than dumping them all on one neighbour, and allows weighting by assigning more positions to more capable hardware.
Hot shards and rebalancing
Even distribution of keys does not mean even distribution of load, because traffic per key is rarely uniform. A shard holding one very active tenant can saturate while others idle. The mitigations are to shard on a composite key that splits the heavy entity, to give the outlier its own dedicated shard, or — for time-series data where the hot range is always "now" — to include something other than time in the leading position so new writes spread.
Rebalancing is the operational reality of sharding, and it is a live-data migration: copy the range while it continues to receive writes, catch up on the changes made during the copy, then cut over reads and writes atomically, verify, and only then delete the source. Systems that do this well make it incremental and resumable, because a rebalance of a large shard takes hours and must survive interruption.
A: Row count is not load. One shard is receiving a disproportionate share of traffic — typically because it holds a single very high-volume tenant, a celebrity account, or the current time range in a time-partitioned scheme where all new writes target the newest partition. Even key distribution says nothing about request distribution, so the metric to look at is operations and bytes per shard, not size. The fixes depend on the cause. If a single large tenant is responsible, the answer is either to give that tenant its own dedicated shard, which is straightforward with directory-based routing, or to change the key so that tenant's data is itself split — a composite of tenant plus something high-cardinality within it, accepting that queries for the whole tenant now fan out. If it is a hot individual entity, the same key-splitting logic applies at entity level. If it is time-based skew, the leading component of the key must not be the timestamp; prefixing with a hash or a bucket spreads new writes across shards, at the cost of making time-range scans a fan-out. In all cases I would check first whether the traffic is legitimate, because a hot shard is also what a misbehaving client or a missing cache looks like, and adding shards to absorb a bug is expensive.
The distributed transaction problem
Once data spans shards, a single operation that must change two shards atomically has no cheap answer. Two-phase commit gives you atomicity but blocks if the coordinator fails at the wrong moment, and it makes availability the product of all participants' availability.
The usual answer in practice is to avoid needing it. Choose the shard key so entities that change together live together — the "aggregate" boundary in domain-driven terms. Where that is impossible, use a saga: a sequence of local transactions each with a compensating action, accepting that the system is temporarily inconsistent and that compensation must be idempotent. And use the outbox pattern to make "write to the database and publish an event" atomic, by writing the event into the same transaction as the data and having a separate process publish it, rather than trying to write to two systems at once.
Quick recall
Short-answer versions of the same material, for spaced repetition.
A: No. Every write is still applied by one primary, and shipping the log to more replicas adds load. Only sharding scales writes.
A: Almost all of them, because the node count is the divisor. Going from ten nodes to eleven reassigns roughly ninety percent of the keyspace.
A: About one over N, because only the arc belonging to the added or removed node changes owner.
A: Even distribution, by giving each physical node many ring positions so arc sizes average out. They also allow weighting larger machines.
A: The dual write. It records the event in the same database transaction as the data, so a crash cannot leave one without the other.
A: A user writes to the primary then reads from a lagging replica and sees their old data, which looks to them like a failed save.