Scaling a database with replication and sharding

One commodity Postgres box handles roughly 5,000 simple queries per second, and as much data as one machine's disk holds. Past that there are two moves: copy the data so more machines can serve reads, or split the data so more machines can absorb writes. Replication and sharding fix different bottlenecks, and the expensive mistake is sharding a system whose write rate would have fitted on a single box for another two years.

Find the bottleneck before choosing the tool

Take a product with 5M daily active users, each doing 30 reads and 2 writes a day. Run the back-of-the-envelope arithmetic:

reads:  5,000,000 x 30 = 150,000,000/day  ÷ 86,400 ≈ 1,700 QPS avg → ~5,200 QPS at 3x peak
writes: 5,000,000 x  2 =  10,000,000/day  ÷ 86,400 ≈   116 QPS avg → ~  350 QPS at 3x peak

The read path sits at the 5,000 QPS ceiling; the write path uses 7% of it. That 15:1 ratio is the ordinary shape of a consumer product, and it says replication, not sharding — sharding here buys write capacity nobody needs and charges cross-shard joins for it. Cheaper than either: check first whether caching absorbs the reads.

Replication: one leader, many followers

Replication keeps full copies of the database on several machines. One node is the leader and takes every write; changes ship as a stream to one or more followers, which serve reads. It buys two things: read throughput, and a machine that can be promoted when the leader dies.

Three followers split the 5,200 QPS peak read load into about 1,730 QPS each, comfortably inside the ceiling, while the leader keeps the 350 write QPS. What replication never buys is write capacity: every follower applies 100% of the write stream, so the twentieth follower is doing exactly as much write work as the leader. When the write rate alone approaches 5,000 QPS, adding nodes stops helping and the answer becomes sharding.

Synchronous costs latency, asynchronous costs data

Synchronous: the leader waits for a follower to acknowledge. In the same datacenter that is one 0.5 ms round trip on a ~5 ms commit — cheap. Across regions it is not: an India-to-US-East round trip is ~200 ms, taking the commit from 5 ms to 205 ms. By Little's Law that is 350 QPS x 0.205 s ≈ 72 writes in flight at once against 350 x 0.005 ≈ 2 before, so a pool of 20 connections is exhausted and the write path stalls.

Asynchronous: the leader confirms immediately and ships changes behind the client's back. Writes stay at 5 ms, but the follower runs behind — call it 50 ms in-region, ~200 ms across an ocean. If the leader dies and a follower is promoted, everything in flight is gone: 350 QPS x 0.2 s ≈ 70 committed writes lost silently, after the client was told they succeeded.

This is CAP in its only useful form. When a partition cuts the leader off from its follower, synchronous replication refuses the write (consistency) and asynchronous accepts it and lets the copies diverge (availability). PACELC covers the other 99.9% of the time: with no partition, synchronous trades latency for consistency, asynchronous the reverse.

The bug lag actually causes

Replica lag is not an abstract "con". It breaks read-your-writes:

Replication lag breaks read-your-writes: a profile update disappears for 200 ms100%−+FitSVGEsc

Scroll to zoom · drag to pan · 0 fits · Esc closes

The write succeeded and the page reloaded showing the old name, so the user submits again — a duplicate write and a support ticket. Two cheap fixes: route a user's reads to the leader for a window longer than the worst observed lag (500 ms covers a 200 ms replica), or have the client carry the log position of its write and make the follower catch up to it first. Choose per read path — a user's own profile needs read-your-writes, someone else's follower count does not.

More than one leader

Multi-leader replication lets two or more nodes accept writes and replicate to each other, which helps when writers sit in different regions. It also creates conflicts: two regions 200 ms apart can each accept a write to the same row inside that window, and now there are two truths. Last-write-wins settles it by discarding one — fine for a "last seen" timestamp, unacceptable for a balance. Unless the data partitions by region or the writes commute, keep one leader; electing one is what consensus protocols are for.

Sharding: splitting the data

Sharding splits one logical database into shards, each holding a subset of the rows on its own machine. Replication copies all the data; sharding divides it. That is what makes it the write-scaling and dataset-size answer.

Suppose the write path grows to 1 billion events/day, or about 12,000 QPS average. Against a 5,000 QPS ceiling, and leaving 50% headroom for peaks and rebuilds, 12,000 ÷ 2,500 ≈ 5 shards, rounded to 8 so the hash space divides evenly. At 1 KB per row, 1 billion rows/day is also 1 TB/day of new data, which settles the question on its own.

Everything then depends on the shard key, which fixes both the distribution and which queries stay cheap.

• Range-based — user IDs 1–1,000 on shard 1, 1,001–2,000 on shard 2. Simple, and range scans stay on one shard. It also invites the classic hotspot: shard on an auto-increment ID or created_at and every new write lands on the newest shard, pushing 12,000 QPS through one 5,000 QPS node while seven machines idle.
• Hash-based — hash the key to pick the shard. Distribution evens out, at the price of range queries, which now fan out to all 8. Use consistent hashing so the map survives resizing.
• Directory-based — a lookup service maps key to shard. Most flexible, and it puts a hop in front of every query: 0.5 ms in-datacenter, on a service that is now a single point of failure. Cache the map in the app process (a 100 ns memory read) and invalidate it when shards move.

Name the costs before an interviewer does. Fan-out inherits the worst shard: if each shard's p99 is 10 ms, a query touching all 8 exceeds 10 ms about 1 - 0.99^8 ≈ 8% of the time, so the tail gets fatter as you add shards. Rebalancing moves bytes: going from 8 shards to 16 with plain hash mod N remaps half the keyspace, and 4 TB over a 1 Gbps link at ~125 MB/s is 4,000,000 MB ÷ 125 ≈ 32,000 s, about 9 hours of migration — consistent hashing moves roughly 1/16 instead. Schema changes run 8 times and must be safe in the mixed state between them.

The two together

Large systems use both. Each shard is a small replicated cluster with its own leader and followers: sharding supplies write throughput, replication supplies availability and read capacity inside each shard.

Two shards by user id, each a leader with an asynchronous follower100%−+FitSVGEsc

Scroll to zoom · drag to pan · 0 fits · Esc closes

The picture has a price: a query spanning users A–M and N–Z now runs in the application, and a read routed to F1 can be 50 ms stale.

In an interview

What is being tested: whether you can separate a read bottleneck from a write bottleneck with arithmetic, and whether you know what each fix costs. Be ready to say when you would use replication (availability, read scaling) versus sharding (write scaling, datasets too large for one machine), and to explain the models and strategies with their trade-offs.

What to say: "reads peak near 5,200 QPS and writes near 350, against roughly 5,000 QPS on one box, so this is a read problem — cache first, then two or three followers. I would shard when writes approach a few thousand QPS or the dataset outgrows one disk." Then name the shard key and the query it makes expensive; volunteering the costs of sharding reads as experience.

The mistake that loses points is reaching for shards with no number attached, on a system doing 350 writes/second. Close behind: adding followers and never mentioning lag — introduce replica reads and in the same breath say that a user who writes and immediately reads may not see their own write, and how you route around it. Third: calling replicas a backup, when a DELETE replicates in milliseconds. Related: horizontal vs vertical scaling and NoSQL stores, most of which ship sharding built in.

Check yourself

A service does 2M reads and 800,000 writes a day, both peaking at 4x average. Replicas, shards, or neither?

2,000,000 ÷ 86,400 ≈ 23 QPS reads and 800,000 ÷ 86,400 ≈ 9 QPS writes; at 4x peak, 93 and 37 QPS. Both sit under 3% of one box's ~5,000 QPS, so neither. Add a follower if availability requires failover, and spend the effort on indexes: sharding buys nothing here and taxes every future join.

Your leader is in Mumbai and you want a follower in Virginia for disaster recovery. Synchronous or asynchronous, and what does the choice cost?

Asynchronous. Synchronous adds a ~200 ms round trip to every commit, taking a 5 ms write to 205 ms, which by Little's Law is 350 x 0.205 ≈ 72 concurrent writes and an exhausted pool. Async costs you the failover window: at ~200 ms lag, promoting Virginia loses roughly 350 x 0.2 ≈ 70 acknowledged writes. If those are payments, keep a synchronous follower in-region (0.5 ms) and the cross-region copy async.

You shard an events table by event_id hash across 8 nodes. The main query is "all events for user 42 in the last 7 days". What breaks, and what would you change?

Every read fans out to all 8 shards for the application to merge, so the query inherits the slowest one: with a 10 ms p99 per shard it misses 10 ms about 8% of the time, and each added shard makes that worse. Shard by user_id instead — one user's events live on one node and the query becomes a single-shard range scan. The trade is that a hot user is now a hot shard, handled by splitting that key's range rather than rehashing everything.