Replication, Sharding & Scaling

A user updates their profile photo, the app says “saved,” they refresh, and the old photo is back. Nobody changed it — the read just went to a replica that hadn’t caught up yet. This is the read-your-own-writes trap, and it’s one of the most common, most confusing bugs in any system that scaled its reads. This module shows you the machinery underneath — replication, sharding, and what CAP actually forces you to choose — so that bug, and the design decisions around it, stop being mysteries.

The Goal

By the end of this module you can:

• Explain primary/replica replication and how it scales reads without scaling writes
• Diagnose replication lag and the read-your-own-writes trap, and name the fixes
• Describe what failover is and why a replica fleet is also your availability story
• Distinguish sharding from partitioning, and explain why the partition-key choice is everything
• Reason about CAP with a concrete during-a-partition example, not the textbook triangle

The Lesson

One database becomes a problem at scale

Your SplitEase API runs on a single PostgreSQL instance. That’s correct for now — one machine handles a surprising amount. But two pressures eventually break it:

• Read load. Ten thousand people opening the app to check balances — all those SELECTs hit one machine.
• Availability. If that one machine dies, the whole app is down. One database is a single point of failure.

The first tool for both is replication.

Replication — copies that follow the leader

Replication means running extra copies of your database that stay in sync with the original. The standard shape is primary/replica (older docs say master/slave):

• The primary is the one database that accepts writes (INSERT, UPDATE, DELETE).
• Every write is streamed to one or more replicas, which apply the same changes.
• Reads can be served by any replica — and that’s the win.

flowchart TD
    App["Application"] -->|writes| P["Primary - accepts writes"]
    App -->|reads| R1["Replica 1 - read only"]
    App -->|reads| R2["Replica 2 - read only"]
    P -->|streams changes| R1
    P -->|streams changes| R2

This scales reads: most apps read far more than they write (checking balances vastly outnumbers creating expenses), so spreading reads across replicas relieves the primary. Add more replicas, serve more reads. PostgreSQL does this with streaming replication off its write-ahead log — the same WAL that gives you durability also ships the changes to replicas.

The catch you must internalise: replication scales reads, not writes. Every replica must apply every write, so adding replicas doesn’t help a write-heavy load at all — they’re all doing the same write work. Writes still bottleneck on the single primary. That limit is exactly what pushes systems toward sharding later.

Replication lag and the read-your-own-writes trap

Replicas apply changes a moment after the primary commits them. That gap — usually milliseconds, occasionally seconds under load — is replication lag. Most of the time it’s invisible. The trap is when it isn’t:

1. A user updates their profile — the write goes to the primary and commits.
2. The app immediately reads the profile to show it back — but that read is load-balanced to a replica.
3. The replica is 200 ms behind and hasn’t applied the update yet.
4. The user sees their old data, one second after the app said “saved.”

This is read-your-own-writes broken. It’s maddening to debug because nothing is actually wrong — the data is correct on the primary, the replica catches up a moment later, and by the time you go looking it’s consistent. The fixes, in order of bluntness:

• Read from the primary after a write. For a short window after a user writes, route that user’s reads to the primary. Simple, common, slightly defeats the read-scaling for that window.
• Stick the session to the primary for a few seconds after any write.
• Wait for the replica to catch up to the write’s position before reading (some systems support this; it adds latency).

The mental model to keep: the moment you have replicas, “the database” is no longer a single source of truth at every instant — it’s a primary plus copies that are slightly behind. Every read-after-write in your app must be designed with that in mind.

Failover — replicas are also your uptime

The second job of replicas: if the primary dies, a replica can be promoted to become the new primary. That’s failover. With it, a hardware failure means a brief blip while a replica takes over, not an outage until someone rebuilds a server.

flowchart TD
    subgraph Before
        P1["Primary - healthy"] --> RA["Replica A"]
        P1 --> RB["Replica B"]
    end
    subgraph After["After primary fails"]
        X["Primary - dead"]
        RA2["Replica A - promoted to primary"] --> RB2["Replica B"]
    end

Failover can be manual (a human promotes a replica) or automatic (a tool detects the dead primary and promotes one). Automatic is faster but has its own danger — split-brain, where a network hiccup makes two nodes both think they’re the primary and both accept writes, which then conflict. That’s why automatic failover needs careful coordination (a quorum agreeing who’s in charge). For now, just hold: a replica fleet gives you read scaling and an availability plan in one investment.

See It Move

Watch the lag indicator and follow one write from the primary to the replicas — then notice what a read sees if it arrives before the replica catches up.

Step through it and notice:

• The write only ever lands on the primary — replicas never accept writes, they receive them.
• There’s a visible delay before each replica reflects the new value — that gap is replication lag.
• A read that hits a lagging replica returns the old value even though the primary already committed the new one — that is the read-your-own-writes trap in one frame.
• More replicas means more read capacity, but each one still has to apply the same write — which is exactly why this scales reads and not writes.

The Lesson, continued

Sharding — splitting the data itself

Replication copies the whole database many times. That’s useless for two problems: a write load too big for one primary, and a dataset too big to fit on one machine. The tool for those is sharding — splitting the data across multiple databases, where each holds a different slice and accepts its own writes.

flowchart TD
    App["Application"] --> Router["Routing by shard key"]
    Router -->|users A to H| S1["Shard 1"]
    Router -->|users I to P| S2["Shard 2"]
    Router -->|users Q to Z| S3["Shard 3"]

Now writes spread out: a write for a user in shard 1 doesn’t touch shards 2 or 3. Three shards, roughly triple the write capacity, and a dataset three times too big for one machine now fits. This is horizontal scaling of the database — covered in system-design terms in scaling distributed systems.

Sharding vs partitioning — the distinction interviewers probe

These get conflated, so be precise:

Partitioning is splitting a big table into smaller pieces within one database (PostgreSQL does this natively — e.g. one partition per month). Sharding is splitting across separate database servers, each its own primary. Rough rule: partitioning is a single-machine organisation trick; sharding is a multi-machine scaling strategy. Sharded databases are often also partitioned and replicated — the techniques stack.

The partition key is everything

The one decision that makes or breaks a sharded system is the partition key (a.k.a. shard key) — the column that decides which shard a row lives on. Get it wrong and you inherit pain you can’t easily undo:

• Hot shards. Shard by country and 70% of your users are in India — one shard melts while the others idle. The key must spread load evenly.
• Cross-shard queries. Shard by user_id, then someone needs “all expenses in this group” where the group’s members live on different shards. Now a single query must hit every shard and merge results — slow, complex, the thing you sharded to avoid.
• Cross-shard transactions. A transaction touching two shards can’t use one database’s ACID guarantee — you’re into distributed-transaction territory, which is genuinely hard.
• You can’t easily change it later. Re-sharding a live system means moving enormous amounts of data while it’s serving traffic.

The skill: choose a key that (a) spreads writes and storage evenly and (b) keeps the queries you run most often inside a single shard. For SplitEase, sharding by group_id would keep “this group’s expenses” on one shard (good) but make “this user across all their groups” cross-shard (a cost you’d accept only if group-scoped reads dominate). There’s rarely a perfect key — it’s a trade you reason about explicitly. And the senior move: don’t shard until you must. Sharding adds permanent complexity; replicas, caching, and a bigger machine buy you a long runway first.

CAP, made concrete

CAP is stated as “pick two of Consistency, Availability, Partition-tolerance,” which sounds like a riddle. Here’s what it actually means in practice.

A partition is a network failure that splits your nodes into groups that can’t talk to each other — say your primary in one data centre loses its link to a replica in another. Partitions will happen; you don’t get to opt out of network failures. So P isn’t really a choice — it’s a fact of distributed life. The real choice is: when a partition happens, do you favour Consistency or Availability?

Make it concrete. A user’s account balance lives on two nodes that just got cut off from each other, and a withdrawal request arrives at one of them:

• Choose Consistency (CP). The node says “I can’t confirm I have the latest balance, so I refuse the request.” The user is blocked, but you never serve a wrong balance or allow a double-spend. A bank picks this — better to reject than to be wrong about money.
• Choose Availability (AP). The node says “I’ll answer with what I have.” The user gets a response, but it might be stale, and two sides of the partition might both allow a withdrawal that conflicts when they reconnect. A social feed picks this — better to show a slightly old like count than an error page.

flowchart TD
    Part["Network partition happens - nodes cannot sync"] --> Q["A request arrives - what do you do"]
    Q --> CP["CP - refuse rather than risk a wrong answer - banks money"]
    Q --> AP["AP - answer with possibly stale data stay up - feeds counters"]

That’s CAP with the abstraction stripped off: during the unavoidable network failure, you either stay correct (and sometimes refuse) or stay available (and sometimes wrong). It’s the same ACID-vs-BASE trade from the previous module, now you can see exactly when the choice gets forced. Relational/ACID systems lean CP; many NoSQL/BASE systems lean AP — and modern ones often let you tune it per query.

Check The Concept

How This Shows Up At Work

• The phantom bug ticket. “User says their edit didn’t save, but it’s fine in the database.” Whoever recognises read-your-own-writes — the read hit a lagging replica — closes a ticket that three people couldn’t reproduce. It’s one of the most common scaled-system bugs.
• The premature shard. A team shards a database at 100k rows “for scale,” inherits cross-shard queries and re-sharding pain, and runs slower than a single well-indexed Postgres would have. Knowing when not to shard is as valuable as knowing how.
• The shard-key postmortem. A company sharded by region, India became 70% of traffic, one shard saturated, and the fix required moving terabytes live. Engineers who can interrogate a proposed shard key for hot spots and cross-shard queries before it ships are gold.
• The interview staple. “How do you scale a database for reads? For writes?” Reads → replicas (and name the lag trap unprompted). Writes → sharding (and name the partition-key problem unprompted). Volunteering the trade-offs, not just the technique, is what separates senior answers.

Build This

You can see real replication concepts locally with two PostgreSQL containers — primary and replica — using Docker. If Docker isn’t set up, read the steps and run the lag reasoning; the understanding is the goal, not the ops.

1. Start a primary configured to allow replication:

docker run -d --name pg-primary -e POSTGRES_PASSWORD=secret -p 5433:5432 postgres:16

2. Connect and write some data to the “primary”:

psql -h localhost -p 5433 -U postgres

CREATE TABLE balance (user_id INT PRIMARY KEY, amount_paise BIGINT);
INSERT INTO balance VALUES (1, 500000);

3. Simulate the read-your-own-writes window in your head with the numbers in front of you. A write commits on the primary at T=0. A replica that’s 200 ms behind serves a read at T=50 ms — it returns the old row. The same read at T=300 ms returns the new one. That 200 ms is the entire bug. Write down: which reads in your SplitEase API happen immediately after a write? (Showing the just-created expense back to the user is the obvious one.)

4. Reason it through — pick a shard key for SplitEase. On paper, suppose you must shard. For each candidate key, write which common query stays single-shard and which goes cross-shard:

   • Shard by user_id: “my expenses” → single shard. “this group’s expenses” (members on different shards) → cross-shard.
   • Shard by group_id: “this group’s expenses” → single shard. “all my groups” → cross-shard.

   There’s no free lunch — you’re choosing which query to make cheap and which to make expensive. That is the design decision.

5. Break it on purpose — the hot-shard thought experiment. Suppose you shard expense by the first letter of the city name. List the busiest Indian cities — a huge share start with M (Mumbai), D (Delhi), B (Bengaluru). Notice three shards would carry most of the load while others sit idle. That’s a hot shard, and it’s why high-cardinality, evenly-distributed keys (like a hash of the user id) usually beat human-meaningful ones. Clean up when done:

docker rm -f pg-primary

Interview Practice

Replication, sharding, and CAP questions asked at Indian backend and fintech interviews. Answer aloud before peeking.

Where to Practice

Check Yourself

1. In primary/replica replication, which node takes writes, and what scales when you add replicas?
2. Why does adding replicas not help a write-heavy workload?
3. What is replication lag, and how does it cause the read-your-own-writes bug?
4. Name two fixes for read-your-own-writes.
5. What is failover, and what is split-brain?
6. Sharding vs partitioning — one line each.
7. Why is the partition-key choice so consequential, and what two properties do you want in a key?
8. State the CAP choice during a partition with a concrete CP and AP example.

Explain it out loud: Explain to an empty chair why a user might see their just-saved profile photo revert for one second — name the primary, the replica, replication lag, and the exact fix — then explain why adding more replicas wouldn’t help if the problem were too many writes instead. If you stumble on “replication scales reads not writes,” that’s the section to re-read.

Why AI Can’t Do This For You

AI can describe primary/replica replication and sharding perfectly — it’s in every system-design article it trained on. What it can’t do is look at your traffic and tell you whether you have a read problem or a write problem, whether that phantom “my edit didn’t save” ticket is replication lag or a real bug, or whether the shard key someone proposed will create a hot shard given your user distribution. Those answers live in your metrics and your data, which the model can’t see.

And the highest-value judgment here is restraint — knowing not to shard yet, to reach for an index and a cache and a replica first, because sharding’s complexity is permanent. That spine comes from understanding the machinery well enough to push back when someone wants to over-engineer for scale they don’t have. No prompt gives you the confidence to say “not yet” in a design review.

Module done? Add it to today’s tracker