Replication, Sharding & Consistency
Copy data for reads, split it for writes, and confront the CAP trade-off. Where consistency models come from.
One database node eventually runs out of something: CPU, disk, memory, write throughput, or patience during maintenance. Distributed data systems use two big moves. Replication copies data to more nodes. Sharding splits data across nodes. Both buy scale, and both introduce consistency questions.
This is the heart of system design: once data lives on more than one machine, “the value” is no longer a single obvious thing at every instant.
The mental model
Replication is copying. Sharding is partitioning.
Replication: every node has a copy
writes -> leader -> follower A
-> follower B
reads -> leader or followers
Sharding: each node owns a slice
user_id hash 0..33 -> shard A
user_id hash 34..66 -> shard B
user_id hash 67..99 -> shard CReplication mainly scales reads and improves availability. Sharding mainly scales writes and storage. Most large systems use both: each shard is itself replicated.
Leader-follower replication
The common pattern is one leader for writes and multiple followers for reads. The leader accepts writes, appends them to a log, and followers replay that log.
Benefits:
- Read traffic spreads across replicas.
- A follower can be promoted if the leader fails.
- Backups and analytics can run against replicas instead of the leader.
Cost:
- Followers can lag behind the leader.
- Failover can lose or replay recent writes if not designed carefully.
- Read routing gets subtle: should this user read from the leader or a replica?
Replication lag is the everyday consistency bug. A user saves a profile, refreshes, and a replica returns the old profile. The usual fix is read-your-writes routing: after a user writes, route that user’s reads to the leader or a sufficiently caught-up replica for a short window.
Sharding strategies
- 1Writeclient → leader
- 2Replicateleader streams log
- 3Followers applyasync or sync
- 4Readsserved from followers
A shard key decides where data lives. The key is a design decision, not an implementation detail.
| Strategy | How it works | Good at | Risk |
|---|---|---|---|
| Hash sharding | hash(key) -> shard |
Even distribution | Range queries scatter across shards |
| Range sharding | key ranges -> shard |
Efficient range scans | Hot ranges and uneven growth |
| Directory sharding | Lookup table maps key to shard | Flexible moves | Directory becomes critical metadata |
| Composite sharding | Tenant plus bucket/time key | Balances locality and spread | More complex queries and rebalancing |
Hot shards happen when one key or range receives disproportionate traffic: celebrity accounts, viral posts, current timestamps, or one large enterprise tenant. A good shard key has high cardinality and spreads the write/read load you actually have.
Rebalancing
Shards are never balanced forever. Users grow, tenants churn, and one shard fills first. Rebalancing strategies include:
- Virtual shards — split data into many logical partitions, then assign those partitions to physical nodes.
- Consistent hashing — adding a node moves only part of the keyspace.
- Range splits — split a hot or large range into smaller ranges.
- Dual writes / backfill / cutover — migrate carefully when changing keys.
The hard part is not moving bytes. It is serving reads and writes correctly while bytes are moving.
CAP and consistency models
CAP says that during a network partition, a distributed system must choose between:
- Consistency — every read sees the latest committed write.
- Availability — every request receives a non-error response.
Partition tolerance is not optional in real networks. The practical question is which features need strong answers during failure and which can tolerate stale answers.
| Model | Promise | Example fit |
|---|---|---|
| Strong consistency | Reads see the latest committed write | Payments, inventory reservation |
| Eventual consistency | Replicas converge if writes stop | Like counts, feeds, analytics |
| Read-your-writes | A user sees their own recent writes | Profile edits, settings |
| Monotonic reads | Once a user sees version N, they do not later see older versions | Session-scoped UX consistency |
Pattern recognition
Variations
Worked problems
For each scenario, state what you are scaling: reads, writes, storage, geography, or availability.
Scale reads without breaking profile edits
A profile service handles 1,000 writes/sec and 80,000 reads/sec. The primary database can safely serve 5,000 reads/sec in addition to writes. Design the read scaling path and handle the case where a user edits their profile then refreshes.
Approach. Use followers for generic reads, but route recent writers around lag. Estimate replica count from peak read capacity, then add headroom.
Show solution
If each read replica can handle about 5,000 reads/sec, then:
80,000 reads/sec / 5,000 per replica = 16 replicas
Add headroom and maintenance capacity: start around 20 replicasRead routing:
write profile -> leader
-> replication log -> followers
read own profile right after write -> leader or caught-up replica
read public profile from others -> any healthy followerAfter a successful write, store a short-lived marker in the session such as profile_written_at. For the next minute, route that user’s profile reads to the leader or to replicas whose replay position is past the write. Everyone else can read from followers.
Monitor replication lag as a first-class metric. If lag exceeds the UX tolerance, fall back to leader reads for sensitive endpoints, shed noncritical reads, or show “changes are saving” instead of pretending stale data is current.
Choose a shard key for chat messages
Design storage for a chat product. Most conversations are tiny, but some rooms have millions of members during live events. Messages are usually read by conversation in time order. What shard key do you choose, and how do you handle hot rooms?
Approach. The primary access pattern is messages by conversation and time, so pure user sharding is wrong. But a single conversation key can become hot, so add bucketing for very large rooms.
Show solution
A baseline key:
partition = hash(conversation_id)
row order = created_at, message_idThis keeps normal conversation reads local and spreads conversations across shards. Range-sharding by created_at alone would send all current live-event writes to the same hot range.
For huge rooms, split one conversation into buckets:
partition = hash(conversation_id, bucket_id)
bucket_id = hash(sender_id) % N # or assigned by room configurationReads for a hot room now fan out across N buckets and merge by timestamp. That is more complex, so only enable it for rooms that need it. Normal rooms should stay simple.
For rebalancing, use virtual shards: many logical partitions mapped onto fewer physical nodes. When adding capacity, move virtual shards gradually and keep a routing table so clients know where each partition lives. During moves, either proxy old-to-new requests or dual-read until the cutover is complete.
Pick consistency for product features
A social commerce app has two features: users can like products, and sellers can reserve inventory during checkout. Pick a consistency model for each and explain what happens during a regional network partition.
Approach. Match the consistency model to user and business harm. Likes are experience data; inventory is a correctness invariant.
Show solution
| Feature | Model | During a partition |
|---|---|---|
| Product likes | Eventual consistency | Accept likes locally, merge counts later, tolerate temporary drift |
| Seller inventory reservation | Strong consistency | Route to the inventory leader/quorum or reject writes until safe |
For likes, availability matters more than exact counts. If two regions both accept likes, the count can converge through an event log. Duplicate events need idempotency keys, but a temporarily stale count is acceptable.
For inventory, overselling is business harm. If the system cannot reach the leader or quorum for the item, it should fail closed: show “try again” rather than promise a reservation it cannot serialize.
A nice middle ground for UX is read-your-writes for the user who clicked like: show their own like immediately from local state, while the global count catches up asynchronously.