Database Sharding and Replication

A single Postgres instance on a db.r6g.16xlarge (64 vCPU, 512 GB RAM) can handle roughly 50k simple reads/sec and ~5k writes/sec before latency climbs past 50 ms. For a mid-size product that number is fine — until it isn't.

Two independent bottlenecks hit at different times:

1. Read throughput: your reporting queries, search, and API reads swamp the CPU. The writes are fine.
2. Write throughput / storage: insert rate exceeds what one disk and WAL pipeline can absorb; or you simply have more data than fits in one machine's storage budget.

The solutions are also independent: replication solves (1); sharding solves (2). Using sharding to solve (1) — when replication would have sufficed — is one of the most common premature optimization mistakes in system design. Reach for read replicas first.

A read replica receives a continuous stream of changes from the primary via the write-ahead log (WAL in Postgres, binlog in MySQL). Reads route to replicas; all writes go to the primary. One primary + three replicas roughly 4x your read capacity with zero application logic changes (just point your read connection pool at a replica DNS endpoint).

The catch: replication lag. Replicas apply changes asynchronously. Typical lag is 10–100 ms; under write bursts it can reach seconds. For a user who just updated their profile and immediately reloads the page, routing that read to a lagging replica shows them their old data.

Mitigations:

• Read-your-writes consistency: route reads to the primary for 1–2 seconds after any write by that user.
• Synchronous replication (Postgres synchronous_commit = on): the primary waits for at least one replica to acknowledge the WAL record before confirming the write. Zero lag, but ~2x write latency.
• Session pinning: pin a session to the primary for the duration of a transaction that mixes reads and writes.

Sharding splits data across multiple independent DB nodes, each owning a subset of rows. Two fundamental strategies:

Range sharding: shard by value ranges of the key. Users A–M on shard 1, N–Z on shard 2. Simple to reason about; range scans are efficient (all UK customers on shard 3). Fatal flaw: hot shards — if user IDs are sequential (auto-increment), all new inserts hit the latest shard. If 80% of your traffic is for users created this month, one shard is on fire.

Hash sharding: shard = hash(user_id) % N. Distributes load uniformly. No hot spots from sequential inserts. The penalty: range queries across shards require scatter-gather (fan out to all N shards, merge results). SELECT * FROM orders WHERE created_at BETWEEN X AND Y hits every shard.

Plain hash(key) % N has a fatal rebalancing problem: when you add a shard, N becomes N+1, and almost every key maps to a new shard. You must move nearly all data.

Consistent hashing places both servers and keys on a conceptual ring (0 to 2^32). Each key is owned by the first server clockwise from hash(key). When you add a server, it claims a segment of the ring from its clockwise neighbor — only ~1/N of keys move, not all of them.

Virtual nodes (vnodes): each physical server owns multiple positions on the ring (e.g., 150 vnodes per server). This makes the load distribution more uniform and allows fine-grained capacity weighting — a beefier shard gets more vnodes.

Cassandra, DynamoDB, Riak, and Redis Cluster all use consistent hashing with vnodes. A 10-shard → 11-shard migration moves ~9% of data rather than ~91%. For a 10 TB dataset at 100 MB/s migration throughput, that's 2.5 hours vs 27 hours.

Hash sharding distributes keys uniformly — but not necessarily load. If you shard by user_id and a celebrity user has 50M followers, every follower action (load their feed, like their post) hits shard 7. Shard 7 runs at 100% CPU; the other 9 idle at 10%. This is the celebrity (or hot-key) problem.

Mitigations in order of increasing complexity:

1. Sub-shard: add a secondary random suffix to the key (celebrity_id:shard_suffix) and read from multiple sub-keys, merging client-side. Writes fan out; reads fan in.
2. Replicated shard: detect hot keys dynamically (monitor per-key QPS), copy the hot shard's data to 3–5 read replicas, and route reads proportionally.
3. Separate hot-key tier: explicitly route known hot entities to a dedicated cluster with higher CPU allocation. Instagram does this for mega-accounts.

The celebrity problem is also why sharding by content_id on a social platform is dangerous. Beyoncé's post ID lands on one shard — and it absorbs the entire Super Bowl half-time spike.

Sharding buys write scale at a steep correctness tax. Two operations that are trivial on a single DB become painful:

Cross-shard joins: SELECT users.name, orders.total FROM users JOIN orders ON users.id = orders.user_id — if users and orders live on different shards, there is no single node that can run this join. Options:

• Denormalize: copy user_name into the orders table. Now the join is gone, but writes must maintain consistency in two places.
• Application-side join: fetch from shard A, fetch from shard B, merge in application code. Works, but two network round trips and O(N*M) memory.
• Global reference tables: small, slowly-changing tables (e.g., currency codes) are replicated to every shard.

Cross-shard transactions: two-phase commit (2PC) works but adds coordinator latency (~5–10 ms) and locks rows across shards during prepare. Under failure, a coordinator crash leaves shards locked. Most teams avoid cross-shard writes by ensuring the shard key co-locates all data belonging to one business entity — e.g., all of a user's orders live on the same shard as the user.

The shard key is the most consequential schema decision you'll make — changing it later requires moving all data.

Good shard keys:

• High cardinality (many distinct values → even distribution)
• Co-locate related data (shard by organization_id so all an org's data lives together)
• Match your most common query pattern (avoid scatter-gather for your hot path)

Bad shard keys:

• Low cardinality (e.g., country: 250 values, most users in 5 countries → hot shards)
• Monotonically increasing (e.g., created_at, auto-increment IDs → hot shard at the write frontier)
• Causes cross-shard lookups for your most frequent query

-- Bad: sharding a multi-tenant SaaS by user_id
-- A query for all tickets in an organization spans every shard
SELECT * FROM tickets WHERE org_id = 42;

-- Good: shard by org_id
-- All org 42 data is on one shard; single-node query
SELECT * FROM tickets WHERE org_id = 42;

Rule: design your shard key around your most critical query, not your data model.

Not all replication is primary → replica. Three patterns worth knowing:

Single-primary (most common): one writable primary, N read replicas. Simple, clear ownership. Failover: promote one replica (Patroni for Postgres, MHA for MySQL). Replica promotion takes 10–30 s; brief write outage.

Multi-primary: any node accepts writes; conflict resolution is your problem. Used by CockroachDB, PlanetScale (Vitess), and MySQL Group Replication. Enables geographic write distribution (users write to the nearest region). Complexity: last-write-wins conflicts, split-brain risk.

Chain replication: primary → middle replica → tail replica. Writes confirmed only when the tail acknowledges. Strong consistency, high write latency, rarely used in practice outside storage systems like CRAQ.

For most web services: single-primary with 2–3 async replicas, one synchronous replica for HA failover, and Patroni for automated promotion. Add sharding only when the primary's write throughput or storage is the proven bottleneck — not before.