Advanced 18 min read System Design

Database Scaling: Replication & Sharding

The database is usually the last tier to scale and the first to become a bottleneck. Read replication splits reads from writes and adds failover; sharding splits the data itself across nodes to scale writes. Both trade consistency and complexity for capacity, and choosing a shard key that distributes data evenly while co-locating related rows is the make-or-break decision.

replicationread replicaleadershardingshard keypartitionreplica lagfailoverconnection poolingpostgres

Prerequisites: Scalability Basics Transactions Postgres Deep Dive

Introduction

What is this?

Replication keeps copies of the database on multiple nodes: a leader accepts writes and streams changes to read replicas, which serve reads and can be promoted during failover. Sharding splits the data set across many independent database nodes so each holds a slice, distributing both storage and write load. Replication solves read scaling and availability; sharding is the tool of last resort that scales writes, at the cost of cross-node queries, transactions and joins that no longer work as before.

Motivation

Why do you need it?

Reads usually dwarf writes; offloading reads to replicas multiplies read capacity without touching the write path.
Replicas provide failover targets, so a leader crash becomes seconds of interruption instead of downtime.
A single database has hard limits on storage, connections and write throughput that no machine upgrade can cross.
Sharding spreads writes across nodes, so throughput scales roughly linearly with the number of shards.
Separate replicas let analytics and reporting queries run without degrading the primary workload.

Think about it

Mental model

Imagine

One librarian guarding the only catalog

Think of

Copy shops holding the catalog, split by topic

Replication is like every branch holding a copy of the same catalog — readers use any branch, but only headquarters accepts edits. Once the catalog outgrows any single building, you split it by topic across buildings (sharding) and tell each reader exactly which building holds the section they need — which also means a single trip can no longer cross-reference another building's shelf.

Mechanics

How it works

The leader records every change in a write-ahead log (WAL) and ships log records to replicas.
Replicas replay the WAL continuously, keeping near-real-time copies; async replicas can lag, sync ones cannot.
The application routes writes to the leader and reads to replicas (or a pooler/proxy does it transparently).
Sharding assigns each row to a node by hashing or ranging the shard key, and a router maps keys to shards.
Operations that span shards (joins, cross-shard transactions, global ordering) must be reworked or removed.
Failover promotes a healthy replica to leader (with or without losing async-durable writes, depending on sync/async).

Writes go to the leader; reads spread across replicas; the data set is split across shards.

ApplicationWrites to leader, reads from replicas

LeaderAccepts writes, appends to WAL

WAL streamChanges shipped to replicas

Read replicasServe reads, candidates for failover

Shard routerhash(shard_key) routes rows to shards

FailoverReplica promoted if the leader fails

Reference

Key commands

docker run -d --name db-leader -e POSTGRES_PASSWORD=secret postgres:16

Start a Postgres leader to experiment against.

Container db-leader started with ID 9a12...

psql -h db-leader -U postgres -c "SELECT pg_create_physical_replication_slot('rep1');"

Create a physical replication slot for a streaming replica.

pg_create_physical_replication_slot\n-------------------------------\n(rep1,)

pg_basebackup -h db-leader -D /var/lib/postgresql/data -R -P -U postgres

Take a base backup of the leader and configure streaming replication.

pg_basebackup: base backup completed

psql -h db-leader -c "SELECT * FROM pg_stat_replication;"

Check the state and lag of connected replicas.

client_addr | state | write_lag\n172.17.0.2  | streaming | 00:00:00.02

psql -c "EXPLAIN ANALYZE SELECT * FROM orders WHERE user_id = 42;"

See whether a query uses an index or scans the table — the first thing to fix before scaling.

psql -c "CREATE INDEX CONCURRENTLY idx_orders_user ON orders(user_id);"

Build an index without blocking writes on the table.

Hands-on

Examples

Read/write splitting and shard selection in a Node.js data layer

// db.js - route writes to the leader and reads to replicas; shard by user id
const { Pool } = require("pg");
 
const leader = new Pool({ host: "db-leader", database: "app" });
const replicas = ["db-replica-1", "db-replica-2"].map(
  (host) => new Pool({ host, database: "app" })
);
 
function hashString(str) {
  let h = 0;
  for (let i = 0; i < str.length; i++) h = (h * 31 + str.charCodeAt(i)) >>> 0;
  return h;
}
 
function readPool(userId) {
  return replicas[hashString(String(userId)) % replicas.length];
}
 
async function getUser(userId) {
  const { rows } = await readPool(userId).query(
    "SELECT * FROM users WHERE id = $1",
    [userId]
  );
  return rows[0];
}
 
async function createUser(user) {
  // Writes always go to the leader; replicas catch up via the WAL.
  return leader.query(
    "INSERT INTO users (id, email) VALUES ($1, $2)",
    [user.id, user.email]
  );
}

Writes flow to the leader only, and reads choose a replica by hashing the user id so the same user consistently reads from the same replica, which avoids read-your-writes surprises in the common case. This is the shard-select pattern in miniature: the same hash function, scaled up to route across shards instead of replicas. In production this routing usually lives in a proxy (PgBouncer/Pgpool, Vitess, or the database client) rather than app code.

Real world

In production

Read replicas serve analytics dashboards and reporting so heavy aggregation queries never degrade the primary transaction path.
Multi-region setups place a read replica near each region so local reads are fast; the leader stays in one region for writes.
SaaS platforms shard by tenant_id so every tenant's data and access pattern stays within one shard, with queries naturally scoped.
Connection pooling (PgBouncer) sits between thousands of app instances and the database to survive connection storms and keep the pool healthy.

Guidelines

Best practices

Add read replicas first and monitor write throughput — only shard when writes, storage or hot shards force you to.
Choose a shard key with high cardinality and even distribution that co-locates data accessed together (user_id for a per-user app).
Route every query for a key to one shard; keep cross-shard work out of the hot path or offload it to a separate read model.
Monitor replica lag and alert on disconnects; treat lag as a first-class availability metric.
Run schema changes with expand-contract and CONCURRENTLY index builds so migrations never block the leader.

Watch out

Common mistakes

Sharding before replication and caching are exhausted, paying for complexity that simpler levers could have solved.
Choosing a hot shard key (country, tenant that is 90% of traffic), leaving one node saturated while the rest idle.
Ignoring replica lag and serving stale reads where freshness matters (payments, balances).
Building cross-shard joins and transactions that were never possible, forcing app rewrites after the shard split.

Root cause analysis

Common issues

Error

Replica lag grows to minutes and reads return stale data

Root cause

Long-running queries or heavy analytics on the replica slow WAL replay, or autovacuum is not keeping up.

Fix

Move heavy queries off the replica, add indexes for the replay workload, and tune autovacuum; consider an async standby with its own resources.

Error

FATAL: sorry, too many clients already

Root cause

The combined connections from app pools exceed max_connections.

Fix

Insert a connection pooler like PgBouncer with transaction-level pooling, and cap each app's pool size.

Error

Shard 3 uses 80% of its disk while shards 1-2 are idle

Root cause

A skewed shard key (few very large tenants) is concentrating data and traffic.

Fix

Split the hot shard or re-key with a composite key (tenant_id + sub_key) so large tenants spread across nodes.

When things break

Debugging tips

Check pg_stat_replication or SHOW REPLICA STATUS for lag and replication state when reads look stale.
Use EXPLAIN ANALYZE to catch sequential scans; a missing index is the most common 'database scaling' symptom.
Watch the connection count — hundreds of app instances opening pools can exhaust max_connections before the DB is actually busy.
When a shard looks hotter than others, query the shard key distribution to find skewed keys before rebalancing.

Interviews

Interview checkpoint

Practice

Hands-on lab

Postgres Primary + Replica in Docker

45 min · advanced

In progress

Objectives

Run a Postgres leader with streaming replication
Verify writes replicate to the replica
Measure replica lag and observe read-your-writes caveats
Promote the replica to leader and confirm failover

Postgres Primary + Replica in Docker

45 min · advanced

0/5 steps

sandbox · Linux sandbox

Try these commands

Test yourself

Quick quiz

0/3

What is the primary job of a shard key?

With asynchronous replication, a committed write is always guaranteed to exist on the replica after a leader failure.

Heavy analytics queries are slowing down the production transaction database. The application itself is well-indexed. What is the first fix?

Quick reference

Cheat sheet

Replication

Leader (primary)Accepts writes, owns the WAL

Read replicaServes reads, catches up via WAL

SynchronousWaits for ack; no data loss, slower

AsynchronousFast; may lose recent writes

FailoverPromote a replica to leader

Replica lagTime the replica is behind the leader

Sharding

Shard keyHigh cardinality, even distribution

Hash shardingEven but loses range locality

Range shardingGood for range scans, risks skew

Cross-shard queryAvoid; rebuild as read model

Re-shardingExpensive; choose the key once

Go deeper

Official documentation

PostgreSQL — Warm Standby & Streaming Replication

docs

PostgreSQL — Failover and Recovery

docs

MySQL Replication

docs

Vitess — Sharding for MySQL at Scale

docs

Database Scaling: Replication & Sharding

What is this?

Why do you need it?

Mental model

How it works

Key commands

Examples

Read/write splitting and shard selection in a Node.js data layer

In production

Best practices

Common mistakes

Common issues

Debugging tips

Interview checkpoint

Hands-on lab

Postgres Primary + Replica in Docker

Quick quiz

Cheat sheet

Related topics

Official documentation