Ch 23 / 24 Scaling Under Real Load 0/0 exercises Exercises ↓

AmouAI Hub/Courses/Software Engineering for AI Engineers/Chapter 23

Part 5 · Scaling and Operating in Production · Chapter 23

Scaling Under Real Load

The distributed system in the architecture diagram is more impressive than the single large machine, and for most products it is also slower, more expensive and considerably harder to debug. This chapter is about knowing when the numbers actually say to distribute — and about the one decision, the shard key, that you cannot change afterwards without moving every row you own.

Reading
DDIA Ch. 5–6 · Vogels
Focus
The key that cannot be changed
Modes
architecture · JS · choice
Exercises
12

1Vertical scaling goes further than the diagrams suggest

Start with the arithmetic, because it is the part that gets skipped. A single cloud instance you can rent today reaches into the hundreds of gigabytes of RAM and a hundred-plus cores, and a managed Postgres on such a machine will comfortably serve tens of thousands of simple queries a second against a working set that fits in memory. Most products — including most products whose teams are drawing sharding diagrams — are nowhere near that. Ledger, at 400,000 users and roughly 40 million requests a month, averages about fifteen requests a second and peaks near sixty. That is not a distributed-systems problem. It is a problem for one machine with a replica for safety.

The reason this matters is not thrift, though it is cheaper. It is that distributing a system introduces a category of failure you did not previously have: partial failure. On one machine, an operation succeeds or fails. Across a network, it can also succeed slowly, succeed and be lost, succeed twice, or leave you unable to find out which. Every one of Chapter 18’s retries, timeouts and idempotency keys exists because of that category, and every one of them is code you must write, test and operate. A team that distributes before it needs to has paid that cost in exchange for nothing.

Vertical scaling has three real limits, and it is worth naming them precisely so you can tell whether you are near one. Machine size is a ceiling, and it is high; you will usually hit a different limit first. Cost curves bend: the largest instances cost disproportionately more per unit of capacity, so somewhere before the ceiling, two medium machines beat one large one on price. And a single machine is a single failure domain — which is the limit that actually binds for most teams, and the reason the first horizontal step is almost always a replica for availability rather than a shard for throughput.

Before scaling anything, exhaust the cheap wins, because they are usually larger than the scaling. Chapter 8’s missing index turns a sequential scan into a lookup and can be a hundredfold. Chapter 7’s cache removes the query entirely. The N+1 that issues 340 queries instead of two is a rewrite of one function. Connection pooling is a configuration line. Teams routinely add machines to a system that is doing a hundred times more work than it needs to, and they get a linear improvement on a problem that had a hundredfold fix sitting in it.

2Stateless services and load balancing

Horizontal scaling of the application tier is easy exactly to the extent that the tier is stateless: any request can go to any instance, because the instance holds nothing that request needs. State goes to the database, the cache, or the object store. When that is true, adding capacity is adding a process, and losing capacity is losing a process, and neither is an event.

What breaks it is almost always the same short list. In-memory sessions, so a user who lands on a different instance is logged out. In-memory rate-limit counters, so the effective limit is the configured one multiplied by the instance count. Local file uploads, so the file exists on one machine and the next request cannot find it. Background schedulers running on every instance, so a nightly job runs six times. Each is easy to fix once named, and each is invisible with one instance — which is why they are all discovered on the day capacity is added.

Then the balancing algorithm, which matters more than its reputation suggests:

  • Round robin distributes requests evenly and assumes requests are equal. When they are not — and with model calls in the mix they are emphatically not — an instance can be handling three eight-second requests while another handles thirty fast ones, and round robin cheerfully sends the next one to whoever is next in line.
  • Least connections sends to whoever is least busy right now, which is the right default for workloads with variable duration. This is most AI workloads.
  • Least response time weights by observed latency; better still, and more sensitive to noisy measurement.
  • Consistent hashing sends the same key to the same instance, which is how you get cache locality — and which reintroduces the hot-key problem from Section 5.
  • Sticky sessions pin a user to an instance. They are a way of keeping state in the tier while claiming not to, and they defeat autoscaling: a new instance receives only new sessions, so the overloaded instances stay overloaded while the fresh one idles.

Two mechanisms make the pool trustworthy. Health checks must exercise the dependencies the service actually needs — a check that returns 200 from a handler with no database access will keep an instance in the pool while every real request fails. And graceful shutdown: on a deploy, the instance should first fail its health check, wait for the balancer to notice, then drain in-flight requests, then exit. Skipping that means every deploy is a small burst of errors, which people learn to ignore — and ignoring a class of error is how a real one goes unnoticed.

3Replication: read scaling, and read-your-writes

A read replica is a copy of the database that receives a stream of changes from the primary and serves reads. It buys two different things that people conflate: availability, because a replica can be promoted when the primary dies, and read capacity, because reads can be spread across replicas while writes still go to one place. Most workloads are read-heavy — Ledger is about 92% reads — so this goes a long way before anything is partitioned.

What it costs is replication lag. The replica is behind the primary by some amount, usually milliseconds and occasionally seconds, and under write bursts or long-running queries considerably more. A read served from a replica is therefore a read of a slightly older world. Most of the time nobody notices. The time somebody notices is precise and predictable: immediately after they wrote something.

The user edits an invoice, the write goes to the primary, the page reloads, the read goes to a replica that is 400 ms behind, and the old value comes back. The user concludes the save failed and does it again. This is read-your-writes, and it is the consistency guarantee people actually expect even though almost nobody states it. Three ways to provide it, in increasing order of subtlety:

  • Route to the primary for a window after a write. Record when this user last wrote, and for the next few seconds send their reads to the primary. Simple, effective, and it needs somewhere to keep that timestamp that every instance can see — usually the session or a small cache.
  • Route by object. Send reads of the specific object the user just wrote to the primary, and everything else to replicas. Narrower, and more code.
  • Wait for the replica to catch up. Record the write position, and have the read wait until the replica has reached it. Correct, and it converts a lag problem into a latency problem.

The mistake to avoid is routing all reads to the primary the moment this bug appears, which works and gives back the read scaling you added the replica for. The window is what makes it cheap: a few seconds of primary reads for the small fraction of users who just wrote something.

4Sharding, and the shard key that decides everything

Sharding splits the data itself across machines so that each holds a subset. It is the step that genuinely raises write throughput and storage beyond one machine, and it is also the step that changes what your database can do — because a query that spans shards is no longer one query, a transaction that spans shards is no longer a transaction, and a join across shards is either a distributed query engine or your own application code.

Everything about how well this works is decided by the shard key: the field whose value picks the shard. Choose it well and each shard takes a similar share of the data and the traffic, and the queries you run most carry the key so they touch exactly one shard. Choose it badly and you have all the operational cost of a distributed database with the throughput of the busiest single machine.

Four properties, in the order they usually bite:

  1. Even distribution of data — no shard holds a disproportionate share of the rows.
  2. Even distribution of traffic, which is a different question. A key can spread rows perfectly and still send 40% of queries to one shard.
  3. Present in the common queries. If your hot query does not carry the shard key, every one of its executions is a scatter-gather across every shard, and its latency becomes the slowest shard’s latency.
  4. Stable. A key whose value changes means the row must move between shards, which is a delete and an insert with no transaction around it.

Two more choices come with sharding and are worth deciding deliberately. Hash versus range: hashing distributes evenly and destroys range queries, since consecutive values land on different shards; ranges preserve locality and concentrate traffic on whichever range is currently hot. And rebalancing: if you shard by hash(key) mod N, then changing N moves almost every row, which for a large table is a migration measured in days. Consistent hashing with virtual nodes moves roughly 1/N of the keys instead, and you will implement exactly that in the exercises.

5Hot keys and what they cost

A hot key is a single value that receives a disproportionate share of the traffic: the enterprise tenant, the celebrity account, the product on the front page, the one document every agent retrieves. Sharding does not help, because every request for that key goes to the same shard by definition — that is what sharding is.

The cost is worth computing rather than describing, because the number is unintuitive. With eight shards and perfectly even traffic, each takes 12.5%. If one key takes 40% of the traffic, its shard is running at over three times the average, so either it falls over or you size every shard for the hot one — and then you are paying for eight machines that could handle 3.2 times your total load, of which you use one shard’s worth on seven of them. Effective capacity is roughly the average divided by the peak, and the exercises make you compute it.

The remedies, in the order to try them:

  • Cache it. A hot key is by definition frequently read, which makes it the ideal cache entry. Often this is the whole answer, and it is Chapter 7’s material arriving exactly where it is most valuable.
  • Replicate it. Keep a copy of the hot key on every shard, if it is read-mostly and can tolerate being slightly stale.
  • Split it. Change the key so the hot value spreads — the composite key above, or salting, where the key becomes tenant_id:bucket with a small random bucket, so one tenant becomes sixteen keys. Reads must then query all sixteen and merge, which is the price.
  • Isolate it. Give the enormous tenant its own shard, or its own database. This looks like giving up, and for a genuine outlier it is frequently the right answer: one customer who is 40% of your data is not really the same workload as the other 399,999.

6Build: Ledger’s scaling plan and its trigger conditions

A scaling plan is not a target architecture. It is a sequence of steps, each with a trigger condition stated as a number, so that the decision to take the next step is made by measurement rather than by whoever is most worried. Here is Ledger’s, written the way it should be written.

Now. One primary Postgres, one synchronous standby, two application instances behind a least-connections balancer. About 15 requests a second average, 60 at peak; the database is at roughly 12% CPU and the working set fits in memory.

Step 1 — add read replicas. Trigger: database CPU above 60% at peak for three consecutive days, or read latency p95 above 150 ms with no missing index to blame. Reads are 92% of traffic, so this is the cheapest large win available. Ships with a five-second primary window after a write and a lag-threshold fallback, because the read-your-writes bug is not hypothetical — it is certain.

Step 2 — move the large tenant to its own database. Trigger: any single tenant exceeding 15% of rows or 15% of query volume. This is the isolation remedy, taken before it becomes a shard key problem, and it is deliberately placed before general sharding because it is far less work and removes the thing that would have made sharding hard.

Step 3 — shard the invoice tables. Trigger: write throughput above 60% of what the primary can sustain, or the largest table past roughly 2 TB. Key: hash(tenant_id, invoice_id), chosen because tenant-scoped queries are the common case and the composite prevents a single tenant from landing on one shard. Consistent hashing with 128 virtual nodes per shard, so adding a shard moves about 1/N of the keys rather than nearly all of them.

Never, unless a number says otherwise. Splitting the application into services for scaling reasons; the tier is stateless and adding an instance is a configuration change. Multi-region; Ledger has no latency requirement that needs it and it would double the operational surface. Both of these are written down as declined with a reason, which is what stops them being re-proposed every quarter.

Checkpoint

Playground

A different dataset: a document store for a retrieval system, where the access pattern is “fetch the chunks for this document” and one reference manual is retrieved by almost every query. Find the key that is even and keeps a document’s chunks together.

Exercise set

Twelve problems: consistent hashing with virtual nodes, a skew metric, resharding cost, read-your-writes routing, weighted round robin, a replica-lag budget, hot-key detection, and two architecture decisions where the data distribution does most of the work. Your progress is saved in this browser.

All Warm-up Core Challenge Reset chapter

Chapter 24 — Operating, Evolving and Paying Down

The last chapter, and the capstone. Incidents, postmortems, debt that is deliberate and debt that is not, and a full design review you defend against the tradeoffs it made.

Continue →