Scaling the Database

Read replicas: scaling reads, and the lag you buy

Start with the cheapest move. One primary database accepts every write and appends it to a log (the write-ahead log in Postgres, the binlog in MySQL). One or more replicas stream that log and replay it, so each holds a copy of the data. Reads can go to any copy; writes still go to the primary. Because most services read far more than they write, this moves most of the query load off the machine that cannot be cloned. Snip's redirects are almost pure reads, so five replicas can serve roughly five times the lookups one primary could.

The price is set by when the primary tells the client the write succeeded.

Most deployments run asynchronous replicas for read scale and, where durability matters, one semi-synchronous replica as the failover target. Fully synchronous replication to every replica is rare: it makes the system as slow and as fragile as its worst member.

Replication lag and read-your-own-writes

With async replication a replica is always some distance behind: typically milliseconds, but seconds during a bulk write, a long-running query on the replica, or a network hiccup. That gap is replication lag, and it produces a specific bug. A user creates a Snip link, the write commits on the primary, the app redirects them to their dashboard, and the dashboard query lands on a replica that has not replayed the insert yet. The link they made one second ago is not there. Nothing is broken; the system is behaving as designed. It violates read-your-own-writes, which is the guarantee users assume.

Three fixes, from blunt to precise:

• Read from the primary after a write. Route reads that follow a write, or reads of data the user just changed, to the primary. Simple and correct, and it keeps replicas for everything else.
• Sticky-to-primary window. After any write, pin that user to the primary for a fixed window, say 5 seconds, longer than your normal p99 lag. A cookie or session field carries the timestamp. It is crude but needs no lag measurement.
• Causal tokens. The write returns the log position it landed at (a Postgres LSN, a MySQL GTID). The client sends it with later reads, and the router sends the read only to a replica that has replayed past that position, or to the primary if none has. This is exact and costs the most plumbing.

Connection pooling: bounding the most expensive thing you can open

A database connection is not a cheap socket. In Postgres each connection is a dedicated backend process with its own memory, commonly several megabytes and much more under heavy queries; MySQL uses a thread per connection, lighter but still real. The server also does work proportional to the connection count: lock tables, snapshot bookkeeping, and the OS scheduling a few hundred runnable processes onto a handful of cores. Throughput does not keep rising with connections. It rises to roughly the core count, then flattens and falls as the processes contend.

The service side makes this worse. Fifty app instances with 20 connections each is 1,000 connections, and most are idle most of the time, since a request holds a connection only while a query runs. A pool such as PgBouncer sits between app and database and multiplexes many client connections onto a small, fixed set of server connections. In transaction mode a server connection is lent out for one transaction and returned.

Size the pool with Little's Law: the number of requests in flight equals arrival rate times time in the system. At Snip's peak of about 8,000 lookups per second that miss the cache, each taking 5 ms in the database, in-flight queries average 8,000 × 0.005 = 40. A pool of 50 to 100 server connections is plenty; 2,000 would only add contention. When all pool slots are busy, new requests wait in a queue instead of piling onto the database.

The single-primary write ceiling

Replicas copy writes; they do not share them. Every replica replays the same log the primary produced, so adding replicas raises read capacity and leaves write capacity exactly where it was. Worse, each replica adds replay work and network fan-out for the primary. Writes scale only by making the primary bigger, or by giving each write a different primary.

Bigger works for a long time: more cores, more RAM so the working set stays in the buffer pool, faster NVMe. Do this first. It is the cheapest move and changes no code. It then runs out in three ways:

• Write throughput. The log is written serially and every write flushes to durable storage. A tuned single node sustains tens of thousands of simple writes per second, not millions, and its failover target must keep up too.
• Dataset beyond one machine. Once the hot data no longer fits in the buffer pool, reads become disk seeks, and replicas have to hold the same full copy, so they do not help.
• Index maintenance. Each insert updates every index. A table with five indexes does roughly six writes per row, and big B-tree indexes that no longer fit in memory turn each one into random I/O. Backups, schema changes and vacuum also lengthen with size.

When you can name which of these you are hitting and the next machine size will not clear it, you partition. Not before: sharding is the most expensive move in this guide, and many systems never need it.

Partitioning and sharding: one table, many databases

Partitioning splits one logical table into pieces. The pieces can live in one database (a Postgres partitioned table, useful for dropping old data cheaply) or on separate servers, and then it is called sharding. Each shard is a full database with its own primary and replicas, holding a slice of the rows. Write capacity now adds up: ten shards take roughly ten times the writes, and each holds a tenth of the data and indexes.

Every row needs a rule that says which shard owns it, built from a shard key, a column in the row. Two ways to turn a key into a shard:

Range partitioning on a timestamp or an auto-increment ID is the classic trap: all inserts hit the last shard while the others sit idle. Hashing scrambles the order and removes that hotspot, at the cost of ordered access. Pick based on how you query, not on how the data looks.

Choosing the shard key

This is the most consequential decision in the whole ladder, because changing it later means moving every row. A good key has four properties:

• High cardinality: many more distinct values than shards, so rows can spread. A country code on 16 shards fails.
• Even load: values are hit about equally. A key that is balanced by row count can still be wildly unbalanced by traffic (see the next section).
• Present in your hottest queries: if the key is in the WHERE clause, the query goes to one shard. If not, it goes to all of them.
• Stable: it never changes for a row, since changing it means moving the row between servers.

Requests then need to find the shard. Either a routing tier (a proxy such as Vitess's VTGate) receives the SQL, extracts the key and forwards it, or the client library holds a directory of shard locations and connects directly. The proxy keeps clients simple and lets you move shards without redeploying apps. The client-side route saves a hop but spreads the shard map into every service. Either way, the map is now critical metadata: if two routers disagree, writes land in the wrong place.

Skew and hot partitions

A hash spreads distinct keys evenly. It does nothing about unequal traffic per key. A single key always hashes to the same shard, so if one key receives a large share of the requests, one shard takes that share however many shards you have. Adding shards does not help the hot one; it is already as parallel as it gets.

Snip has the textbook case. A link goes viral and draws 50,000 redirects per second, more than the whole system averages. All of it lands on one shard, whose CPU and connection pool saturate, so every other link that shares the shard slows down with it. The same applies to the click counter for that link: thousands of increments per second on one row, all serialized by the row lock.

Mitigations, by where the pressure falls:

• Absorb reads upstream. A hot read is one value requested constantly, which is the best case for a cache. A short-TTL entry in the cache tier or at the edge takes the 50,000 rps before it reaches the shard.
• Split hot keys (salting). For hot writes, append a small random suffix: code#0 … code#7. Writes spread over eight rows (and so, via hash, up to eight shards); a read of the total sums all eight. Apply it only to keys you have detected as hot, since the read fan-out is the cost.
• Separate the counter path. The click count does not need to live in the links row. Buffer increments elsewhere and apply them in batches, so a link's lookup and its counting never compete. That design is the topic of Async Work & Queues.

The costs you inherit

Sharding does not make the database faster; it makes each query smaller, as long as the query stays on one shard. The moment it cannot, you pay.

• Cross-shard joins have no single engine to run them. A router must fetch from each shard and join in memory, or the app must do it. Latency becomes that of the slowest shard, and a fan-out to 16 shards makes tail latency far more likely.
• Cross-shard transactions need a protocol such as two-phase commit, in which a coordinator asks every shard to prepare and then to commit. It holds locks across network round trips and blocks if the coordinator fails. Many systems simply do not offer it.
• Resharding is operational work. With hash(key) mod N, changing N from 16 to 17 moves about 16/17 of all rows. Consistent hashing instead places shards (and keys) on a ring so that adding one shard moves only about 1/N of the keys. The common practical form is virtual nodes: create many logical shards up front (say 1,024), map them onto the physical servers, and rebalance by moving whole logical shards. Even so, you copy live data while it takes writes, then cut over.

So design to avoid all three. Denormalize so that the data a query needs lives with its shard key: copy the owner's name onto their rows rather than joining a users table on another shard. Choose a shard key so that the transactions you need are within one shard. Treat any cross-shard transaction as a design smell, and replace it with an idempotent, retryable step where you can. Anything that truly needs global consistency, such as a unique custom alias across all shards, goes to a small, separate, unsharded store.

Snip, concretely

Start with the honest question: do Snip's numbers force sharding? The links table is 1B rows × ~500 B ≈ 500 GB. That fits on a large single server, and link creation is only about 1M a day, around 12 writes per second. Taken alone, a primary plus replicas, a pool and a cache would carry it for a long time. Three things push past that: the table keeps growing at roughly 500 MB a day with indexes on top; the buffer pool cannot hold 500 GB of hot data on a sensible instance; and the click stream, 100M events a day at peaks of 10x the average, writes at thousands per second, which no single primary should share with the link lookup. Those, not the link inserts, justify partitioning.

The design:

1. Shard links by hash(code). The code is high cardinality, immutable, and in the WHERE clause of the redirect query, so SELECT url FROM links WHERE code = $1 goes to exactly one shard as a primary-key point lookup. Using 1,024 logical shards on 16 physical servers gives about 62M rows, around 31 GB, per server, small enough to sit mostly in memory.
2. Each shard has its own replicas for redirect reads, with the user's own just-created link read from the primary for the sticky window described above.
3. A router pool in front of each shard group keeps connections bounded: tens per shard, not thousands.
4. A viral link is absorbed by the cache tier first; whatever misses is a single in-memory row on one shard's replicas.
5. User dashboards ("all my links") are a different access path. Keep a second table keyed by user_id holding code and title: a deliberate denormalization, so that query is also single-shard.

What this does not yet answer is the click path. Counting clicks into the shard that holds the link would put 1,200 to 12,000 writes per second on the same primaries that serve lookups, and hot links on single rows. The next steps are to buffer clicks through queues and aggregate counters, covered in Async Work & Queues, and to decide which reads may be stale and which may not, covered in Consistency & Its Price. For where the click data finally lands for analysis, see the data-side Data Systems Design guide.

✓ Check yourself

Answer from memory before opening each one.

1. A user creates a Snip link and the next page load says “link not found”. Reloading a second later works. Trace exactly what happened, and give two fixes with different costs.

   Show answer

   The insert committed on the primary, but the read went to an async replica that had not yet replayed it: replication lag, a read-your-own-writes failure. Cheap fix: read from the primary for a window (for example 5 seconds) after the user writes. Precise fix: return the log position with the write and route reads only to replicas past it. See Read replicas.

2. A colleague raises the app's pool from 100 to 2,000 connections because requests are queuing, and the database gets slower. Why does it get slower, and what would you check first?

   Show answer

   Each connection is a backend process with memory and scheduling cost; throughput peaks near the core count and then falls from contention. Queuing at the pool was a symptom, usually of slow queries. Compute in-flight work with Little's Law (rate × query time) to see the pool you need, then find why queries are slow. See Connection pooling.

3. Your primary runs at 90% CPU on writes with five replicas. A manager proposes adding five more replicas. What do you tell them, and what do you propose?

   Show answer

   Replicas replay every write, so they add no write capacity and slightly more load on the primary. Options: a larger primary if you have not maxed it out, removing indexes, batching writes, moving write-heavy data (such as clicks) off this database, and sharding if the writes truly exceed one node. See The write ceiling.

4. You range-partition a table by created_at across four servers. A month later one server is at 100% and three are idle. Why, and what changes?

   Show answer

   New rows always have the newest timestamp, so every insert lands in the last range. Hash the key to spread inserts, accepting that time-range queries now touch all shards, or keep ranges for data you query by time and delete by dropping old partitions. See Partitioning.

5. Snip is sharded by hash(code) across 16 shards, yet one shard has 10x the CPU of the others. Rows are evenly spread. What is happening and what are your options?

   Show answer

   A hot key: one viral code takes a large share of traffic, and all of it hashes to one shard. Even rows do not mean even load. Cache the hot reads, salt hot write keys across several rows, and move the counter off the lookup path. More shards would not help. See Skew and hot partitions.

6. Product wants “top 10 links by clicks this week across all users”, and the clicks are sharded by link code. Why is this hard, and how would you serve it without a 16-shard query on every page view?

   Show answer

   The answer needs data from every shard, so latency is the slowest shard and it grows with shard count. Precompute it: each shard reports its own top 10 periodically (the global top 10 is among the shard tops), merge them into a small table or cache and serve that, accepting a delay of minutes. See The costs you inherit.

Takeaway

• Read replicas scale reads only; every replica replays all writes, so they never raise write capacity.
• Asynchronous replication is fast but lags, so reads that must see a user's own write go to the primary, or are pinned or token-routed to a replica that has caught up.
• A database connection is a costly process or thread; a bounded pool sized by Little's Law keeps a slowdown from becoming an outage.
• Scale the primary up first. Shard when write throughput, dataset size or index cost exceeds what the largest sensible machine can carry.
• The shard key is the hardest decision to undo: it needs high cardinality, even load, presence in your hottest queries, and stability.
• Hashing spreads keys but not traffic: one viral key still overloads one shard, and needs caching, key splitting and a separate counter path.
• Sharding makes cross-shard joins, transactions and resharding expensive, so denormalize to keep queries on one shard.

Sharding and replicas give you several copies of the truth that disagree for a moment. How much disagreement a product can tolerate, and what it costs to remove it, is next. → Consistency & Its Price

Further reading

• Designing Data-Intensive Applications, chapters 5 (Replication) and 6 (Partitioning) — Martin Kleppmann, O'Reilly, 2017. The standard treatment of replication lag, read-your-writes, hot spots and rebalancing; read it for the failure cases.
• Vitess documentation — The Vitess project. How a production routing tier, vindexes (shard-key functions) and resharding work for MySQL.
• How Discord Stores Billions of Messages — Discord Engineering, 2017. A real-world sharding story: choosing the shard key (channel plus time bucket), keeping queries single-shard, and why that key is the decision you cannot easily undo.
• PostgreSQL: Log-Shipping Standby Servers — PostgreSQL documentation. Exact semantics of async, synchronous and replay-lag settings.
• PgBouncer features — PgBouncer documentation. What session, transaction and statement pooling allow and break.