Scaling Writes

Chapter 1: Vertical vs Horizontal

The instinctive first move against a 1,000× gap is to buy a bigger box. It is worth taking that instinct seriously and costing it out by hand, in real dollars, rather than dismissing it — because it genuinely helps, and understanding exactly how much and where it stops helping is what tells you when to stop reaching for your cloud provider's console and start reaching for the rest of this lesson.

What a bigger box actually buys a write path

Chapter 0 established that CPU is not the bottleneck for pure single-row commits — the fsync is. So the lever a bigger instance pulls that actually matters here is not more cores, it is faster, more local storage: a bigger instance tier often ships with faster NVMe rather than network-attached block storage, and NVMe's fsync latency is meaningfully lower. That is a real, physical improvement, not a marketing number, and it is worth pricing out precisely.

†Ballpark figures for a managed relational database instance at these specs, rounded to illustrate the shape of the curve — exact pricing varies by cloud provider, region, and reserved-vs-on-demand terms.

Doing the arithmetic on what that buys, per dollar

Compare the Small tier and the X-Large tier directly. The price went up by:

And the ceiling went up by:

Thirteen times the money for four times the capacity. That is not a rounding error or a badly-chosen example — it is the honest shape of vertical scaling on a write path bound by a single serialized durability operation. Cost per additional 1,000 writes/sec makes the divergence starker still:

Each step up the tier ladder costs more, not less, per unit of write capacity gained — the opposite of the volume discount you would expect from almost anything else you buy in bulk. That inversion is the signature of a resource with a hard physical floor (fsync latency cannot go meaningfully below the physics of the storage medium) being approached by throwing money at diminishing returns.

Why vertical scaling cannot close a 1,000× gap on its own

Push the extrapolation to its logical end. If cost kept climbing at even the same 13×-for-4× ratio, reaching 1,000,000 writes/sec from the 1,000/sec floor by vertical scaling alone would require:

No commercially available single database instance offers anything close to a 1,000× improvement in fsync latency over network SSD — that would mean a fsync latency of about one microsecond, faster than the speed of light allows for a round trip to any physical storage medium sitting outside the CPU die itself. Vertical scaling is a real, useful tool for closing a 2–4× gap. It is not a tool that exists for a 1,000× gap, at any price.

The absolute ceiling, at any price

It is worth asking the extreme version of the vertical question: forget cost entirely, what is the best fsync latency any single storage device can offer today, and what ceiling does that imply? The fastest commercially available NVMe drives, attached directly to the CPU with no network hop, report sustained fsync round trips down around 0.05–0.1ms under favorable conditions:

Thirteen thousand. Even spending without limit, on hardware at the edge of what exists, one leader's serialized WAL does not cross five figures of writes per second. Against a 1,000,000 writes/sec target, that best-possible-money-can-buy number is still 75× short. This is the cleanest way to see that the 1,000× gap from Chapter 0 is not a budget problem at all — it is a problem of shape, and no amount of vertical spending changes the shape.

The single point of failure, priced in dollars

Vertical scaling's other cost is not monthly and does not show up on a cloud bill — it is the risk concentrated in having exactly one thing that must not go down. Put a number on it. Suppose a single leader, however large, fails unexpectedly once every few months, and failover to a promoted standby takes a realistic 90 seconds — detecting the failure, promoting a replica, and re-pointing traffic. During those 90 seconds, at even a modest 3,000 writes/sec of real traffic:

At this ad network's $0.002 average tracked value per event, that is roughly $540 of billing data either lost outright (if producers do not buffer and retry) or arriving late enough to complicate reconciliation with the advertiser. One outage. Ninety seconds. A number that scales directly with how much traffic funnels through the single point of failure — which is exactly why horizontal architectures, where losing one of a thousand shards costs 1/1,000th of capacity rather than all of it, are valued for more than their raw throughput math.

What horizontal scaling costs instead

The alternative is not to make the one leader faster, but to have many leaders, each independently absorbing a slice of the traffic — the idea Chapter 2 formalizes as sharding. Cost this the same honest way: N boxes of the cheapest Small tier, each independently capable of 1,000 writes/sec:

For the full 1,000,000/sec target, before any of the batching or sharding refinements the rest of this lesson adds:

That is a real, achievable number today, with nothing more exotic than a thousand ordinary database instances — expensive, and operationally heavy at a thousand boxes, but it is mathematically reachable, which vertical scaling alone is not. And critically, cost here scales linearly with capacity: doubling the boxes doubles both the cost and the ceiling, every time, with no diminishing-returns curve to fight.

The cost horizontal scaling does not put on the invoice

That $620,000/month figure is real, but it is not the whole bill. A thousand independent leaders means a thousand things that can each individually fail, need patching, need backup verification, and need someone paged when one misbehaves at 3 a.m. Put a rough number on that, too — say each additional shard costs an extra 2 hours a month of engineering time to operate at this scale (monitoring dashboards, capacity review, occasional firefighting, its share of on-call), at a loaded engineering cost of roughly $150 an hour:

Add that to the $620,000 of raw compute and the true cost of the naive 1,000-shard horizontal approach is closer to $920,000/month — a number nearly half again the sticker price, and the honest reason teams do not reach for a thousand shards the moment vertical scaling runs out. This is precisely the gap Chapter 5's batching arithmetic closes: fewer, larger shards, each doing meaningfully more work per fsync, cuts both the compute bill and the operational headcount this table implies by roughly the same factor.

A realistic growth path, priced at each stage

Real systems rarely jump straight from one box to a thousand. A concrete, honest trajectory for a growing ad network looks more like this:

That fourth row's near-million-dollar figure is not a typo, and it is not this lesson's final answer — it is the honest cost of solving the problem with only the two tools this chapter has introduced so far. It exists in this table specifically so that Chapter 5's improvement has a real number to be measured against, rather than an abstract claim that batching “helps.”

Notice the crossover: vertical scaling is the right call for the first two stages, and it would be over-engineering to introduce sharding at 500 or even 2,200 writes/sec — the operational overhead this section just costed out is not worth paying until the ceiling a single well-specified box can reach is genuinely, not hypothetically, in sight.

The rule of thumb this table implies: reach for vertical scaling whenever the target is within roughly 2–4× of your current ceiling and a bigger single box comfortably clears it with margin to spare. Reach for horizontal scaling the moment the target is an order of magnitude or more beyond what any realistic single box offers — not because vertical scaling stops working at some magic threshold, but because past that point its cost curve has already bent sharply enough that horizontal's straight line has overtaken it, exactly as the $/1,000-writes-per-second table earlier in this chapter showed.

Neither column is the wrong answer in isolation, and treating this as a one-time either/or decision is itself a mistake — real systems use both, at different points in their growth, and often simultaneously at different tiers of the same architecture (a large vertical leader per shard, in a horizontally sharded fleet, is a completely ordinary production shape). A production write path typically runs the biggest single-leader box that comfortably clears its current peak with margin (vertical, cheap, simple), and reaches for sharding only once the target genuinely exceeds what one well-specified box can do (horizontal, expensive to operate, but the only path that reaches 1,000,000/sec at all). The rest of this lesson takes horizontal as given and asks how to do it well — and Chapter 5's batching arithmetic will cut that $620,000/month, 1,000-box number down by roughly two orders of magnitude before this lesson is done.

Why the same trick worked so well for reads and does not here

This asymmetry is worth naming directly, because it is easy to walk from the companion lesson's Chapter 0 into this one and expect vertical scaling to behave the same way twice. It does not, and the reason is mechanical, not incidental.

Doubling a box's cores roughly doubles its read ceiling, because each core independently serves its own queries in parallel — there is no shared serialization point reads must funnel through. Doubling a box's cores does essentially nothing to its write ceiling, because every write, no matter which core's connection accepted it, must still funnel through the one WAL and the one fsync lane behind it. Adding cores to a write-bound box is like adding more cashiers to a store where every customer, regardless of which cashier rang them up, has to walk through the same single door to leave — more cashiers does not widen the door.

This is also the precise reason the read lesson's replicas do nothing for writes, stated in Chapter 0's quiz: a replica is another copy of the data that can independently answer read questions in parallel, but only the leader's one WAL can accept the writes that keep every copy honest in the first place. More copies of a bottleneck are not the same as a wider bottleneck.

Concept → realization: the same fsync, on faster silicon

It is worth seeing what “faster storage” means at the level of an actual system call, because “fsync latency” can otherwise feel like an abstract knob rather than a real, measurable cost:

strace-style trace, network SSD (Small tier)
fsync(7)                                = 0   <0.9820ms>    ← durable, ~1ms

strace-style trace, local NVMe (X-Large tier)
fsync(7)                                = 0   <0.2410ms>    ← durable, ~0.25ms — 4× faster call, same syscall

Same system call, same semantics, same guarantee — the client waits until the kernel confirms the write is on stable media either way. The only thing that changed is how long the underlying hardware takes to make that true, and that is exactly why the improvement from better hardware is bounded: you are paying for physics to happen faster, and physics has a floor.

It is tempting, staring at that 1ms number, to just turn fsync off — Postgres exposes exactly that setting, fsync = off, and it genuinely removes the bottleneck this entire chapter has been measuring. It also removes the guarantee that made the number meaningful in the first place: with fsync disabled, a commit is acknowledged the instant it hits an in-memory buffer, and an ordinary process crash or power loss can erase any writes that had not yet made it to disk through the operating system's own page cache flush, on its own schedule, with no coordination with your application's promises to its clients. For an ad network being paid against these events, that trade is never worth it — the entire point of Chapter 0 was that a write's value is in its durability, not its speed alone.

Chapter 2: Sharding

Chapter 1 landed on a conclusion, not a design: closing a 1,000× gap means many independent leaders, not one bigger leader. This chapter makes that concrete. A shard is a slice of your data, owned end-to-end by its own leader, with its own WAL and its own fsync lane, completely independent of every other shard's. Sharding is the act of splitting one dataset across many such slices so that write load splits with it. Nothing about a shard is exotic in isolation — it is simply Chapter 0's single box, deployed 1,000 times, each copy responsible for a disjoint fraction of the data instead of all of it.

How many shards, in raw numbers

The arithmetic is the same division Chapter 1 used, now framed as a design decision rather than a cost projection. Take the honest, unbatched per-shard ceiling from Chapter 0 — one leader, one fsync lane, 1ms per commit:

One thousand independent databases, each responsible for its own slice of the click stream, each capable of exactly the 1,000 writes/sec Chapter 0 derived. That number is large and uncomfortable on purpose — it is this lesson's honest floor before Chapter 5 introduces batching, and keeping it visible here is what makes Chapter 5's improvement land as a real, measured number rather than an abstract promise.

It is also worth checking this formula's sensitivity the same way Chapter 0 checked its own: a per-shard ceiling that turns out to be 800 instead of 1,000, a 20% miss on the estimate, changes the shard count from 1,000 to 1,250 — a 25% swing in a number that directly drives how many independent leaders get provisioned and paid for. Measuring the real per-shard ceiling on representative hardware, rather than trusting the back-of-envelope 1,000 figure all the way into a purchase order, is the difference between a plan and a guess.

What determines which shard a write goes to

Every write needs a rule that says, deterministically, which of the 1,000 shards it belongs to — the same rule at write time and at read time, or data silently becomes unfindable. Two families of rule dominate real systems, and they make opposite tradeoffs between how evenly load spreads and how cheaply related data can be queried together.

Hash sharding: spread first, ask questions later

Hash sharding runs a hash function over some field of the write — a click ID, a user ID — and uses the result to pick a shard, typically the hash value modulo the shard count:

A good hash function scatters its outputs close to uniformly across its range, so as long as the keys themselves are reasonably diverse, this rule spreads writes evenly across all N shards almost by construction — no manual tuning, no watching for one shard growing faster than the others under normal conditions. The cost is that a query needing many keys at once — “every click for this campaign, sorted by time” — can no longer be answered by asking one shard. The campaign's clicks are scattered uniformly across all 1,000 shards, by design, and answering that query means asking all 1,000 and merging the results, a pattern called scatter-gather.

Range sharding: keep related data together

Range sharding instead assigns contiguous ranges of a sort key to each shard — shard 1 owns click IDs 0 through 999,999, shard 2 owns 1,000,000 through 1,999,999, and so on, or more usefully for this workload, shard by time: shard 1 owns today's clicks, shard 2 owns yesterday's. A query for “everything in the last hour” touches exactly one shard, cheaply, with no scatter-gather. That locality is the entire appeal of range sharding, and it comes at a cost this workload exposes immediately.

The trap: a time-ranged shard recreates the exact wall this lesson exists to solve

Walk through what happens if this ad network range-shards its click events by timestamp, the seemingly natural choice for a stream of time-ordered events. Every new click, at every moment, belongs to the range containing right now — which lives on exactly one shard, the most recent one. Every one of the 1,000,000 writes a second arriving this instant lands on that single shard, because by definition they all share (nearly) the same timestamp:

That is not a 1,000-way split of the load. It is, precisely, the single-leader write wall from Chapter 0, rebuilt one layer up, wearing a sharding architecture as a disguise. The other 999 shards sit nearly idle, holding yesterday's and last week's data, while the one shard anyone is actually writing to melts exactly the way the single Postgres box did in Chapter 0's opening scene. This is called a hot partition or hot shard, and naive time-based range sharding is the single most common way real systems accidentally build one. It is also one of the most common sharding mistakes in production, precisely because a time-ranged scheme looks obviously correct at design time — the ranges are equal-sized, the schema is clean — and the problem only becomes visible once real, live traffic starts arriving and every single write finds itself pointed at the same one shard, every second.

A hybrid that keeps both properties

Production systems rarely pick one family in isolation. A common pattern for exactly this kind of workload is hash-then-range: hash on something that spreads evenly — say, click ID or a hash of the campaign and a salt — to pick one of the 1,000 shards, and within each shard, store rows ordered by time so that a query scoped to one shard (“this campaign's clicks in the last hour, on the shards it lives on”) is still a cheap ordered scan rather than a full table search. This buys hash sharding's even write spread and most of range sharding's query locality, at the cost of range-across-everything queries still needing scatter-gather — a tradeoff this ad network's billing and analytics use cases accept happily in exchange for never rebuilding Chapter 0's wall by accident.

Pricing the scatter-gather cost, by hand

Hash sharding's cost is not free just because it is described in one clause. Put a number on it. A finance analyst asks for the top 100 clicks by spend for one campaign, sorted by time. Under pure hash sharding, that campaign's clicks are spread uniformly across all 1,000 shards, so answering the query means asking all 1,000:

Parallel fan-out rescues the latency, but not the cost: it still opens 1,000 connections and runs 1,000 queries for a request that, under the hash-then-range hybrid described above, would touch a small, bounded number of shards instead. This is the honest price of hash sharding's even spread — not paid on the write path, where it shines, but on any read that needs to reassemble one logical entity's data scattered by design across every shard.

What this looks like in real systems

None of this is hypothetical — it is the design space every production sharding system lives in, choosing a specific point on the same hash-versus-range spectrum:

Citus's co-location feature is worth a specific mention, because it is a third answer to the tradeoff this chapter has been building toward: rather than choosing hash or range for one table, deliberately shard two related tables — campaigns and their click events, say — by the same key, so that a campaign and all of its clicks always land on the same physical shard, even though the sharding scheme underneath is hash-based and spreads different campaigns evenly. It captures hash sharding's even spread across campaigns and range sharding's locality within one campaign, at the cost of only ever being able to co-locate along one chosen key.

Resharding: the cost of getting the count wrong

1,000 shards is a design-time estimate, and estimates are wrong. Traffic grows past what 1,000 shards were sized for, and the count has to change — which raises an operational question this chapter previews and Chapter 4 answers precisely: when the shard count changes from N to N+1, how much existing data has to physically move to a different shard?

With the naive hash(key) mod N rule above, the answer is almost all of it, and it is worth deriving why rather than taking that on faith. A hash function's output looks, for practical purposes, uniformly random across its full range. Whether hash(key) mod 1,000 and hash(key) mod 1,001 land on the same shard number depends on the specific value of hash(key), and for a uniformly random hash, the two moduli agree only for a small, coincidental fraction of possible values — a back-of-envelope estimate puts the fraction of keys that keep the same shard number at roughly 1÷1,001, meaning:

Adding a single shard to a 1,000-shard cluster, under this naive rule, means physically copying almost the entire dataset — every table, every row, on every one of the original 1,000 shards — to new destinations, while the click stream keeps arriving at 1,000,000 writes a second throughout. That is not a maintenance window, it is a multi-day, high-risk migration that has to keep the lights on the entire time it runs. Chapter 4 builds the specific technique — consistent hashing — that gets that ~99.9% figure down to roughly 0.1%: only the new shard's fair share of keys moves, and nothing else does. That three-orders-of-magnitude difference is not a minor tuning improvement; it is the difference between an operation an on-call engineer can run confidently during business hours and one that needs a dedicated migration project with its own rollback plan, run by a team that has scheduled downtime windows and briefed every dependent service in advance.

A second lever: more logical shards than physical machines

Consistent hashing is not the only answer real systems reach for. A complementary, widely used trick is to decide the shard count once, generously, up front — say 4,096 logical shards — and initially place several logical shards on each physical machine, far fewer machines than logical shards. Growing capacity then means moving whole, already-defined logical shards from a crowded machine to a new one, never recomputing which logical shard a key belongs to at all:

The shard_for(key) function above never changes — it always maps a key to one of the fixed 4,096 logical shards. Only a much smaller routing table, logical shard number to physical machine, needs updating when capacity changes, and moving a logical shard means copying one bounded, known slice of data rather than recomputing a hash function's output for the entire dataset. This buys nearly the same operational benefit as consistent hashing, at the cost of choosing the logical shard count correctly up front — too few, and you eventually hit the same problem this section just derived; too many, and each logical shard is needlessly small. Chapter 4 builds consistent hashing because it removes even that up-front guess, but this fixed-logical-shard pattern is common enough in production (MongoDB's chunk-based sharding is a well-known example) that it is worth recognizing on sight.

Concept → realization: what a shard actually is, physically

It is worth being concrete about what “1,000 shards” means as deployed infrastructure, not just as a number in a formula:

a shard, physically
shard-0442:
  role:        independent Postgres leader (its own WAL, its own fsync lane)
  owns:        rows where shard_key routes to 442 of 1,000
  replicas:    2 followers, for read scaling and failover (Chapter 1's SPOF fix, applied per-shard)
  connects to: an application-tier router that knows the shard_key → shard_id mapping

Every one of those 1,000 leaders needs its own monitoring, its own backup schedule, its own failover plan — the operational overhead Chapter 1 costed at roughly $300,000/month at this scale is not an abstraction, it is exactly this: 1,000 of the box in that code block, each needing the same operational care one box would. Multiply anything that used to be a one-time task — a schema migration, a backup restore drill, a version upgrade — by 1,000, and that multiplication is the real, ongoing cost sharding imposes underneath the throughput win.

What the application has to know

Sharding is invisible to end users and highly visible to application code. Every write, and every read, now needs a routing step before it can reach the right leader:

python
def shard_for(click_id):
    h = hash(click_id)
    return h % NUM_SHARDS

def write_click(event):
    shard_id = shard_for(event.click_id)
    conn = shard_connections[shard_id]      # route to the right leader
    conn.execute("INSERT INTO ad_events ...", event)

def get_click(click_id):
    shard_id = shard_for(click_id)          # the SAME rule, every time, forever
    conn = shard_connections[shard_id]
    return conn.execute("SELECT * FROM ad_events WHERE click_id = %s", click_id)

The read path's get_click function has to call the exact same shard_for logic the write path used, or a perfectly valid click ID becomes permanently unfindable — not because the data was lost, but because the read is asking the wrong one of 1,000 leaders for it. This is the single most common class of bug in a freshly sharded system: a write path and a read path that drift out of sync on shard-assignment logic after one of them gets updated without the other.

That shard_for function is the single most important piece of code in a sharded system: every write, every point read, and every query planner decision downstream depends on it agreeing with itself, forever, across every service that ever touches this data. Chapter 4's consistent hashing exists specifically to let that function's behavior change gradually, a little at a time, instead of needing every caller updated atomically the moment N changes.

That agreement has to hold across every process that ever calls it, which in practice means shard_for cannot simply live as a local function baked into each service's binary — a deploy that updates the function in one service before another would let two parts of the system disagree about where a given key lives, silently, for however long the deploy takes to finish rolling out. Production systems instead centralize this as a shard directory: a small, highly available piece of shared state (often itself just a well-replicated key-value store, or a dedicated routing tier like Vitess's VTGate) that every service consults, so the mapping changes in one place and every caller sees the update at roughly the same time. Getting this directory wrong — serving a stale mapping during a resharding operation — is precisely how writes end up landing on the wrong shard, invisible to any query that correctly asks the new, current shard for them.

Chapter 3: Partition Keys

Chapter 2 ended on a warning: a sharding scheme can be entirely correct — hash based, evenly distributing keys in principle — and still produce a melted shard if the specific key chosen to hash is itself lopsided. This chapter makes that failure mode precise, gives it a name, and derives exactly how bad it gets, by hand, from a real distribution shape ad traffic actually follows. The scheme was never broken; the input to it was. Everything that follows in this chapter is about telling the two apart before production does it for you.

The tempting, wrong choice: shard by campaign

This ad network's obvious partition key is campaign_id — every click belongs to a campaign, campaign-scoped queries (spend reports, fraud checks, billing reconciliation) are extremely common, and shard = hash(campaign_id) mod 1,000 looks, at a glance, exactly like the correct hash-sharding rule Chapter 2 recommended. The problem is not the formula. It is what happens when the values you feed it are not evenly popular.

Ad traffic is not evenly popular — it follows a power law

Click volume across campaigns on a real ad network is famously lopsided: a small number of campaigns — a national brand's product launch, a viral creative — draw enormously more traffic than a typical campaign, and the shape of that lopsidedness has a name and a formula. Zipf's law says that if you rank items by popularity from most to least popular, the frequency of the item at rank k is proportional to 1/ks, for some skew parameter s (s=1 is the classic case, observed across word frequency in language, city populations, and, empirically, ad campaign click volume).

To turn that proportionality into an actual fraction of traffic, normalize by the sum of 1/k across every rank from 1 to N — a quantity with its own name, the N-th harmonic number, written H_N:

And the fraction of all traffic the single most popular item (rank 1) draws, under Zipf with s=1, is simply:

Deriving the top campaign's share, by hand

This ad network runs, at any given time, roughly 12,000 active campaigns. The harmonic number H_N does not have a simple closed form, but it has an excellent approximation for large N, using the natural logarithm and the Euler–Mascheroni constant γ ≈ 0.5772:

Work it through digit by digit. First, the logarithm, split for hand-calculation using ln(12,000) = ln(12) + ln(1,000):

Add the Euler–Mascheroni constant:

And invert to get the top campaign's fraction of all traffic:

One campaign, out of twelve thousand, draws roughly ten percent of the entire platform's traffic. Not because anything is misconfigured — this is simply what a Zipf distribution with a realistic N looks like, and it is the honest shape of most real popularity data, not a pathological edge case invented for this lesson. Any system that partitions by a field correlated with real-world popularity inherits this shape whether or not anyone deliberately designed for it.

What that 10% does to a single shard

Apply the target traffic from Chapter 0. Total platform load is 1,000,000 writes/sec; the top campaign's share is 10%:

Under shard = hash(campaign_id) mod 1,000, every single one of those 100,000 writes a second hashes to the same shard — a hash function is deterministic, and this campaign's ID is one fixed value, so it always lands in the same place. Compare against the honest, unbatched per-shard ceiling from Chapter 0:

One thousand shards, correctly sized on average for 1,000 writes/sec each, and one of them is sitting at a hundred times its capacity while the other 999 comfortably serve the rest of the platform. This is a hot key, and it is the single most common way a technically-correct hash-sharding scheme still melts down in production — the average load per shard looks perfectly healthy on any dashboard that reports fleet-wide averages, which is exactly why hot keys are so often discovered in an incident rather than a design review.

It is not just the top campaign

The same formula gives the second, third, and further ranks, and the picture only gets worse as you look at the shape of the whole distribution rather than a single number:

Every one of those top five, individually, would melt whichever single shard it happens to hash onto, and there is no guarantee the hash function scatters them onto five different shards rather than, by unlucky coincidence, onto fewer. This is exactly the shape of problem that made social-media “celebrity accounts” famous in system-design circles: a small number of outlier keys carry disproportionate weight, and any scheme that maps one logical key to exactly one physical shard is structurally exposed to it.

Comparing candidate partition keys side by side

Before reaching for a fix, it is worth laying out the actual candidates a real design review would consider for this table, and being honest about what each one costs:

The pattern worth internalizing: cardinality and skew are not the same axis, and a key can be high-cardinality and still catastrophically skewed — ad_creative_id above has sixteen times the cardinality of campaign_id and is still worse, because a small number of creatives get reused across many high-traffic campaigns at once. Counting distinct values never tells you whether they are evenly used, and a schema review that stops at “how many distinct campaign IDs do we have” without also asking “how is traffic distributed across them” will walk straight past this problem.

Detecting a hot key before it melts a shard

Every derivation in this chapter is something you can, and should, verify against live traffic rather than trust as a one-time estimate. The concrete signal is per-shard write QPS, watched continuously: 999 shards sitting comfortably near the 1,000/sec design target and one shard consistently pegged at 100,000/sec is not a subtle pattern — it shows up immediately on even the crudest per-shard dashboard, well before that shard's queue depth (the same unbounded growth from Chapter 0, now localized to one shard) starts producing timeouts.

Fix 1: key on something that has no reason to be skewed

The cleanest fix is to stop sharding by the skewed field at all. click_id — a globally unique identifier generated fresh for every single event — has no popularity distribution whatsoever; by construction, every click has exactly one click_id, used exactly once, so hashing on it spreads load as evenly as the hash function itself is uniform, completely independent of how skewed campaigns happen to be:

The tradeoff, previewed in Chapter 2: a query like “every click for campaign 44012, sorted by time” now requires scatter-gather across all 1,000 shards, because clicks for that one campaign are deliberately spread everywhere. For this ad network's write-heavy, analytics-reads-later workload, that tradeoff is usually worth it — the write path, which must survive 1,000,000/sec in real time, gets perfectly even load; the read path, which can tolerate a slower, batched, or asynchronous query pattern, absorbs the scatter-gather cost.

Fix 2: keep the good key, salt away the skew

Sometimes campaign-scoped locality genuinely matters for the read pattern, and giving it up is not acceptable. The standard fix, widely used in systems like DynamoDB under the name write sharding, is to keep campaign_id in the key but append a small random or round-robin salt, spreading one logical campaign's writes across several physical buckets instead of one:

Size B from the arithmetic already on the page. The hottest campaign needs its 100,000 writes/sec spread thin enough that each bucket lands under the per-shard ceiling:

A hundred buckets for one campaign is a lot — and it is worth noticing this number will shrink dramatically once Chapter 5's batching raises the per-shard ceiling from 1,000/sec to 100,000/sec: at that ceiling, the same hot campaign needs only 100,000 ÷ 100,000 = 1 bucket, i.e. no salting at all. Salting and batching are solving overlapping problems, and a system with generous per-shard capacity needs far less of the former.

Salting is not free even when sized correctly: a query for “this campaign's clicks” now has to know how many buckets exist and gather from all of them — a bounded, small scatter-gather (100 shards, not 1,000) rather than the eliminated locality of a single-shard query, but far cheaper than Fix 1's full 1,000-way fan-out for this specific hot key, while cold, unpopular campaigns can use B=1 (no salting at all) since they were never the problem.

A third option: let the system adapt automatically

Both fixes above require a human to notice a hot key and decide what to do about it. Some managed systems try to automate part of this. DynamoDB's adaptive capacity, for instance, monitors per-partition traffic and can transparently isolate an unusually hot partition onto its own dedicated storage node, giving it more of the underlying hardware's throughput without the application changing its key scheme at all. This is a genuinely useful safety net — but it is bounded by the same physics as everything else in this lesson: it can rebalance where a hot key's load lands, not manufacture write capacity that does not exist, and a key skewed severely enough (recall the top-5-campaigns table drawing nearly a quarter of all traffic combined) can still outrun what adaptive rebalancing alone can absorb. Treat automatic mitigation as a safety margin on top of a partition key chosen well, not a substitute for choosing one well. The arithmetic in this chapter is what tells you, in advance, whether you are relying on that margin or actually inside it.

Tuning Fix 2's bucket count as traffic changes

The salt-bucket count derived earlier, B ≥ 100 for the hottest campaign, is a snapshot, not a constant. A campaign's popularity moves over its lifetime — a new launch ramps from cold to viral over hours, then decays over days — and a fixed bucket count is either wasteful early (spreading modest traffic across 100 buckets nobody needs yet) or insufficient at peak (if the campaign outgrows its provisioned buckets). Production systems that lean on this pattern typically recompute bucket counts on a rolling window of recent traffic, the same per-key QPS signal from the detection section above, and adjust the salt range gradually rather than instantly — an instant jump in bucket count for one key is itself a small version of the resharding problem from Chapter 2, since reads for that key now need to know the new range.

Concept → realization: write-sharding a hot key in practice

python
HOT_CAMPAIGNS = {44012: 100, 51188: 64}   # campaign_id → salt bucket count, tuned per key
DEFAULT_BUCKETS = 1                                     # cold campaigns need no salting

def shard_key_for_write(event):
    buckets = HOT_CAMPAIGNS.get(event.campaign_id, DEFAULT_BUCKETS)
    salt = random.randint(0, buckets - 1) if buckets > 1 else 0
    return f"{event.campaign_id}#{salt}"

def read_campaign(campaign_id):
    buckets = HOT_CAMPAIGNS.get(campaign_id, DEFAULT_BUCKETS)
    results = []
    for salt in range(buckets):                    # scatter-gather, but only across THIS campaign's buckets
        key = f"{campaign_id}#{salt}"
        results.extend(query_shard(shard_for(key)))
    return merge_sorted(results)

Notice what this pattern requires that Fix 1 did not: a live, maintained table of which keys are hot enough to need salting, and by how much — HOT_CAMPAIGNS above is not static, it has to track real traffic and adapt as which campaigns are viral changes week to week. That operational cost is the real price of Fix 2's better read locality, and it is a legitimate reason many systems default to Fix 1 (key on something inherently unskewed) unless a specific, durable query pattern justifies the extra bookkeeping. A stale entry in that table — a campaign that was hot last month and is cold now, still carrying B=100 — is harmless, just wasteful; a missing entry for a campaign that just went viral is the dangerous direction to get wrong, since it means no salting is happening exactly when it is needed most, which is also exactly when a hot key is most expensive to discover for the first time.

Either fix assumes the shard count itself is fixed while you choose a key. In practice the two decisions interact: a well-chosen key makes the shard count from Chapter 2 an honest, achievable target, while a poorly-chosen one means no shard count fully protects you, because it is not the number of shards that failed, it is the rule that decides which one a write goes to. Chapter 4 returns to shard count itself, and to the specific pain of changing it, with the tool that makes doing so cheap — and, as a bonus, the same tool improves how evenly an already-good key's load lands across shards in the first place.

Chapter 4: Consistent Hashing

Chapter 2 derived a specific, painful number: adding one shard to a 1,000-shard cluster under naive hash(key) mod N remaps roughly 99.9% of all keys. This chapter builds the technique that gets that number down to about 0.1%, from first principles, by changing not the hash function but the geometry the hash output is mapped onto. The fix does not touch the hash function used elsewhere in this lesson at all — the same hash(key) from Chapters 2 and 3 is reused unchanged; only what happens to its output afterward changes.

The idea: hash onto a ring, not into a bucket

Instead of taking a hash value and reducing it modulo the current shard count, consistent hashing takes a hash value and treats it as a point on a fixed circle — a ring — spanning a huge, unchanging range, typically 0 to 232−1. Both keys and shards get hashed onto this same ring. A key belongs to whichever shard's point is the first one encountered walking clockwise from the key's point.

Nothing in that lookup ever referenced the total shard count N. That single fact is the entire reason this scheme's resharding costs so much less — the assignment rule has no mod N in it to invalidate when N changes, and a rule that never mentions N cannot be disrupted by N changing.

What happens when a shard is added

Add a new shard, shard-C, and hash it onto the ring at some position — say, 2.3 billion, landing between shard-A and shard-B. Only the keys that used to walk clockwise past shard-C's new position on their way to shard-B are affected: they now stop at shard-C instead. Every key that would have landed on shard-A, or that was already past shard-B before reaching shard-C's new position, is completely untouched — its clockwise walk never crosses the new point.

Derive the fraction that moves, honestly, the same way Chapter 2 derived the naive figure. With N shards placed roughly uniformly around the ring, each owns roughly an equal 1/N arc of it. Adding shard number N+1 claims a new arc that is, on average, a fair 1/(N+1) share of the ring's full circumference — and every key in that specific arc, and only that arc, moves to the new shard:

For the same 1,000-to-1,001 shard addition Chapter 2 costed at ~99.9% under the naive scheme:

Compare directly:

A thousand-fold reduction, for the same real-world operation: one more shard added to the same 1,000-shard cluster. This is not a minor tuning gain — it is the difference between a routine, low-risk capacity addition and a multi-day migration project, and it is the reason teams operating clusters at this scale can grow capacity incrementally, a shard or a handful at a time, instead of batching growth into rare, high-risk, all-hands migration events.

What happens when a shard is removed, or fails

The same derivation runs in reverse, and it is worth checking it explicitly because a shard failure is the case that matters most operationally — it is unplanned, and it happens under load, not during a scheduled maintenance window. Remove shard-C from the ring, and every key that used to belong to it, with no code change or manual intervention required, now walks clockwise past its old position to whichever shard comes next — shard-B, in the earlier example. No other shard's territory changes at all:

Compare this against what a single-leader-per-everything design (no sharding at all) would mean for the same failure: 100% of traffic stops, because there was only ever one place for it to go. Consistent hashing turns “a shard died” from a catastrophic, all-traffic event into a bounded, 0.1%-of-keys event — the remaining 999 shards absorb a small, predictable bump in load (each picking up roughly 1/999th more territory) rather than the system losing all capacity at once. This is the same SPOF argument from Chapter 1, now derived with an exact number instead of a qualitative claim, and it is the concrete mechanism by which horizontal scaling's failure-isolation advantage over a single leader actually gets realized in a live system rather than remaining a design-review talking point.

The cost of virtual nodes: memory and lookup time

150 virtual points per shard is not free. At 1,000 physical shards, the ring holds:

Each entry is a hash value (4–8 bytes) plus a shard identifier reference (a few more bytes) — call it 24 bytes per entry as a round working figure:

Memory is not the constraint. Lookup time is the more interesting cost, and it is logarithmic, not linear, because the ring is a sorted structure searched by binary search:

Seventeen comparisons, worst case, to route any key to its shard — a cost so small relative to the millisecond-scale network round trip that follows it that virtual node count can be tuned almost entirely for load-balance quality, not lookup speed. Doubling virtual nodes to 300 per shard only adds one more comparison (log₂ grows by exactly 1 each time the entry count doubles), while meaningfully tightening the load-balance guarantee from the law-of-large-numbers argument above. This is a genuinely rare case in system design where a parameter can be pushed generously in the direction that helps without meaningfully paying for it elsewhere — the honest limiting factor on virtual node count is implementation complexity and the modest memory footprint, not lookup latency.

The problem this creates, and virtual nodes as the fix

A ring with only 1,000 real points on it, placed by hashing 1,000 shard identifiers, does not actually divide the ring into 1,000 perfectly equal arcs — a hash function's output is uniformly random, not evenly spaced, so by chance some shards end up owning noticeably larger arcs than others, purely from randomness in where their single point happened to land. A shard that unluckily owns 3× the average arc length gets roughly 3× the average write load, for no reason related to any hot key from Chapter 3 — simply bad luck in one random hash placement.

The fix is virtual nodes: instead of hashing each physical shard onto the ring once, hash it onto the ring many times, under many different labels — “shard-A-0,” “shard-A-1,” …, “shard-A-149” for, say, 150 virtual points per physical shard. Each virtual point independently claims its own small arc, and a physical shard's total territory is the sum of all 150 of its scattered arcs — and averaging over many independent random placements is exactly the situation the law of large numbers covers: individual unlucky (or lucky) arcs mostly cancel out, and the sum converges toward each physical shard's fair 1/N share.

What actually moving the data looks like, step by step

Knowing that only 0.1% of keys need to move is only half the operation — the other half is moving them without ever answering a read incorrectly or dropping a write mid-flight. The standard sequence, whether the underlying assignment scheme is a ring or anything else, has four steps:

The 0.1% figure this chapter derived is what makes step 2's backfill fast and cheap — a small, bounded amount of data to copy, rather than the near-entire-dataset copy the naive scheme demanded. Steps 3 and 4 exist regardless of which assignment scheme is underneath; they are the general pattern for moving live data without a maintenance window, and consistent hashing's contribution is making step 2 small enough that this whole sequence finishes in minutes rather than days.

Concept → realization: the ring, as a real data structure

A consistent-hashing ring is, underneath the circle metaphor, nothing more exotic than a sorted array and a binary search:

python
import bisect, hashlib

class Ring:
    def __init__(self, virtual_nodes=150):
        self.points = []              # sorted list of (hash_value, shard_id)
        self.vn = virtual_nodes

    def _h(self, s):
        return int(hashlib.md5(s.encode()).hexdigest(), 16) % (2**32)

    def add_shard(self, shard_id):
        for i in range(self.vn):
            h = self._h(f"{shard_id}#{i}")
            bisect.insort(self.points, (h, shard_id))   # O(log n) insert into a sorted array

    def shard_for(self, key):
        h = self._h(key)
        i = bisect.bisect_right(self.points, (h, chr(0x10FFFF)))
        if i == len(self.points): i = 0       # wrap around the ring
        return self.points[i][1]

That is the whole mechanism — a sorted array of hash values, a binary search for the first entry at or past a key's hash, wrapping around to the start if the key's hash was past the last shard's point. Adding a shard means inserting 150 new entries into this array; nothing about any other shard's entries has to change, which is the code-level reason this scheme's data movement is bounded to just the new shard's fair share.

What consistent hashing does NOT fix

It is worth being precise about the boundary of what this chapter solved, because it is easy to walk away thinking consistent hashing also fixes Chapter 3's hot-key problem — it does not, and understanding why sharpens both chapters. Consistent hashing balances how evenly shards divide up the ring's territory. It has no opinion about how evenly traffic is spread across the key space that gets hashed onto that ring. A single hot campaign_id, from Chapter 3, still hashes to exactly one point on the ring and still lands on exactly one shard — virtual nodes make that shard's share of the key space fair, but do nothing about one key inside that share carrying 100,000 writes/sec on its own. The two problems are orthogonal, and a system that only solves one of them has solved half of the real hot-shard problem while leaving the other half fully intact.

What this means for the resharding operation from Chapter 2

Return to the concrete resharding scenario Chapter 2 flagged: growing from 1,000 to 1,001 shards while the click stream keeps arriving. Under consistent hashing with virtual nodes, that operation becomes: hash the new shard's 150 virtual points onto the ring, identify the ≈0.1% of keys whose clockwise-nearest point just changed, and copy only that data to the new shard while it comes online — a bounded, well-understood, low-risk operation instead of the multi-day, nearly-full-dataset migration the naive scheme required. This is precisely why real distributed data stores — Cassandra, DynamoDB's underlying partitioning, Riak, Amazon's original Dynamo paper this technique traces back to — build their partitioning layer on some form of consistent hashing rather than a raw modulus. None of them reinvented the idea from scratch; consistent hashing traces directly to a specific 1997 paper on web-cache load balancing, and its adoption into distributed storage systems is a textbook case of a technique migrating from the problem it was invented for into a much larger one that turned out to share the same underlying shape.

A related technique worth knowing: rendezvous hashing

Consistent hashing is not the only way to get “minimal movement on membership change.” Rendezvous hashing (also called highest random weight hashing) takes a different approach to the same goal: for a given key, compute a combined hash of the key and each candidate shard's identifier, and assign the key to whichever shard produces the highest combined hash value. Adding or removing a shard only changes which candidate produces the highest value for keys that specifically involved that shard in the comparison — giving the same minimal-movement property as the ring, without needing to maintain a sorted ring structure at all, at the cost of an O(N) scan across all shards for every single lookup instead of the ring's O(log N) binary search. For a cluster of 1,000 shards, that is 1,000 hash computations per key routed versus the ring's ~17 comparisons — a real cost that makes the ring the more common choice at this scale, though rendezvous hashing remains popular in smaller clusters (a handful to a few dozen nodes) where its simpler implementation outweighs the lookup cost difference.

This lesson uses the ring for the rest of its examples, because the 1,000-shard cluster this lesson has been building toward sits squarely in the range where its O(log N) lookup cost starts to matter relative to rendezvous hashing's O(N) scan.

Chapter 5: Write Buffering

Every chapter so far has divided the 1,000× gap across more leaders. This chapter closes most of it a completely different way: not adding more leaders, but making each existing leader's single fsync cover far more than one write. This is the improvement every earlier chapter has been referencing forward to, and it is worth deriving carefully, because it is the single biggest number in this lesson.

The arithmetic: amortizing one fsync across many writes

Chapter 0 established the serial ceiling: one commit, one fsync, 1ms, 1,000 commits/sec. The fsync's cost is almost entirely in the physical act of forcing bytes to durable storage — and that cost barely changes whether the buffer being flushed holds one write or a hundred. Group several writes into a single WAL append, fsync once for the whole group, and the fsync cost is paid once but covers every write in the batch:

At a batch of 100 writes per fsync, still 1ms per fsync:

One hundred thousand. The same leader, the same disk, the same fsync latency — a 100× improvement in throughput, purchased entirely by changing how many writes share each fsync call. Nothing about the hardware changed; only the batching policy did.

It is worth sitting with how large that multiplier is relative to everything else in this lesson. Chapter 1's best-case vertical scaling bought a 4× ceiling improvement for 13× the money. This chapter's batching buys a 100× improvement for effectively free — no new hardware, no new shards, just a different policy for when an existing fsync call happens to fire.

Revisiting every number this lesson has built so far

This is the payoff Chapter 0 flagged and Chapter 2 built the shard-count formula to make visible. Re-run both with the batched ceiling:

A hundred-fold reduction in shard count, cost, and operational surface area, from one policy change. This is why Chapter 0 called batching the layer that closes “most of” the 1,000× gap: sharding (Chapters 2–4) turns an impossible problem into a merely large one, and batching turns that large one into something a small team operates comfortably. Ten shards, unlike a thousand, is a fleet size where a single on-call engineer can hold the entire system's shape in their head.

Why you cannot just batch on the client and call it done

The arithmetic above assumes 100 writes conveniently arrive at once, ready to batch. Real traffic does not arrive that way — individual ad-click events land one at a time, from millions of independent client requests, each expecting some kind of acknowledgment. Something has to sit between “writes arriving one at a time” and “writes committed 100 at a time,” accumulating them, and that something is a queue — in production, typically Kafka or a managed equivalent like SQS — placed in front of the database.

This decoupling matters for a reason beyond convenience: it lets the database-facing side batch on its own schedule, independent of exactly when each individual event happened to arrive, while still giving producers a fast, low-latency place to hand off their event.

The batch-size versus latency tradeoff

A batch of 100 does not assemble itself instantly. At this ad network's 1,000,000 writes/sec average, filling a 100-write batch takes:

At full, sustained peak load, waiting for a batch to fill costs almost nothing — a tenth of a millisecond is negligible next to the network round trip the event already traveled. The real cost shows up during quiet periods. At a slow 500 writes/sec (say, 3 a.m. for this country), filling the same 100-write batch by count alone takes:

Two hundred milliseconds of added latency on every single write, at low traffic, is not acceptable for most applications — the fix is a timeout alongside the count: flush whichever comes first, 100 writes accumulated, or a fixed maximum wait (say 10ms) elapsed. At 500 writes/sec, the 10ms timer fires first, flushing whatever partial batch (5 writes, on average) has accumulated:

The system adapts automatically: at high load, batches fill by count and latency stays low; at low load, batches flush by timeout, capping added latency at the timeout value while still buying whatever amortization the traffic naturally supports. Neither extreme requires manual retuning as traffic shifts between them over the course of a day.

Choosing a batch size deliberately, not by guessing 100

The 100-write batch used throughout this chapter was not arbitrary, but it was also not the only valid choice — it is worth deriving how to pick one from an actual latency budget rather than copying a round number. Suppose this ad network's SLA allows at most 20ms of added latency from batching, at the lowest sustained traffic level the system needs to support gracefully, say 2,000 writes/sec overnight. The batch size that exactly consumes that latency budget, by count, at that traffic level is:

A batch size of 40, not 100, is what a 20ms SLA at 2,000 writes/sec actually supports. Check what that buys at peak, 1,000,000 writes/sec:

This is the real tradeoff a design review has to make explicit: a larger batch size buys fewer shards and lower cost, at the price of a larger worst-case added-latency at low traffic. There is no batch size that is simply “correct” — there is only a batch size that matches a stated latency SLA and a stated low-traffic floor, derived the way this section just did, rather than picked because it looked like a clean number.

Batching in real systems, and where the same idea shows up

This exact pattern — buffer, then flush by count or timeout, whichever comes first — recurs constantly once you know to look for it:

Recognizing this as one general pattern, rather than a Postgres-specific trick, is what lets it transfer directly to whichever storage or messaging system a particular design actually uses, and it is the difference between understanding the underlying idea and merely memorizing one tool's configuration flag.

Backpressure: what happens when producers outrun the batch writer

A queue does not create capacity; it only buys time. If producers sustain a rate the batching writer genuinely cannot keep up with — a burst well past 1,000,000/sec, or a temporary slowdown on the database side — the queue itself starts growing, exactly like Chapter 0's unbuffered backlog, just with one extra layer of indirection:

A queue is not infinite. Eventually it fills, and the system needs an explicit policy for what happens next, called backpressure: either producers are told to slow down (the queue exposes a signal, and well-behaved producers respect it), or, past some threshold, excess writes are deliberately dropped — load shedding — rather than let the queue grow without bound and eventually run out of memory or disk entirely.

For this ad network, load shedding on a billing-critical event stream is a last resort, not a default — every dropped click is a real advertiser being under-billed. The honest design target is sizing the queue and the shard fleet generously enough, using this chapter's arithmetic, that backpressure and shedding are rare safety nets rather than routine behavior.

Sizing the queue itself, from a worked overload scenario

Put a concrete number on backpressure rather than leaving it abstract. Suppose a traffic spike — a breaking sports result driving a burst of ad impressions — pushes arrivals to 1,300,000 writes/sec for 90 seconds, against a provisioned batched capacity of 1,000,000/sec across the 10-shard fleet:

At roughly 200 bytes per click event, that backlog is:

5.4GB is a perfectly reasonable amount of data for a properly provisioned Kafka cluster to absorb temporarily — which is precisely the point of sizing the queue generously: a 90-second, 30%-over-capacity spike should be a non-event, quietly absorbed and drained over the next few minutes once the spike passes, rather than a page at 2 a.m. Sizing this buffer from a real, derived number (5.4GB) rather than an arbitrary default is what turns backpressure policy from a guess into an engineering decision.

Concept → realization: a batching writer, end to end

python
class BatchWriter:
    def __init__(self, max_batch=100, max_wait_ms=10):
        self.buffer = []
        self.max_batch = max_batch
        self.max_wait_ms = max_wait_ms

    def add(self, event):
        self.buffer.append(event)
        if len(self.buffer) >= self.max_batch:
            self.flush()                              # count-triggered flush

    def on_timer(self):                             # called every max_wait_ms
        if self.buffer:
            self.flush()                              # timeout-triggered flush, whatever is buffered

    def flush(self):
        batch, self.buffer = self.buffer, []
        conn.execute_many("INSERT INTO ad_events VALUES ...", batch)   # ONE fsync for the whole batch

Two triggers, one shared flush path: whichever condition is met first empties the buffer with a single batched insert and a single fsync. This is, in essence, exactly what Chapter 0's group commit does opportunistically and automatically inside Postgres — this chapter's contribution is making that behavior deliberate, application-controlled, and tuned to this specific workload's shape rather than left to chance. Chapter 0 called group commit opportunistic precisely because it depends on transactions happening to arrive close together in time; this explicit buffer removes that dependency entirely, batching is guaranteed rather than hoped for.

Sensitivity: does the 100× figure hold on real hardware

The 100× improvement this chapter leans on assumed a clean 1ms fsync and a batch of exactly 100. Check it against the fsync-latency range Chapter 0 already tabulated, holding batch size fixed at 100:

Even the most conservative row here — cross-AZ synchronous replication, the safest and slowest option — still needs only 25 shards, a small fraction of the 1,000 the unbatched figure demanded. The qualitative conclusion of this chapter is robust across the entire realistic hardware range: batching is worth doing regardless of which exact storage tier ends up underneath it, even though the precise shard count is worth re-deriving once real hardware numbers are measured rather than assumed.

What batching costs: durability window and partial failure

Batching is not free of tradeoffs. A write sitting in the application-side buffer, not yet flushed, is not yet durable — if the batch-writer process crashes before flushing, those buffered writes are lost, unless the buffer itself is backed by the durable queue from earlier in this chapter rather than pure in-memory state. And a batch insert can fail partway (one bad row among 100), which needs its own handling — typically, retry the batch with the bad row isolated, rather than lose the other 99 good writes alongside it. Neither problem is difficult, but both need explicit handling; batching trades raw throughput for a small amount of added engineering care around partial failure. That trade — a hundred-fold throughput gain for a modest, well-understood amount of extra buffer and retry logic — is, in the context of the numbers this chapter has derived, an easy one to make.

Chapter 6: Multi-Leader Writes

Every shard so far has had exactly one leader. That is correct within a shard, but this ad network serves clients on multiple continents, and a client in Tokyo writing to a leader in Virginia pays a real, physical cost this chapter quantifies before deciding whether to accept it or design around it. Everything built through Chapter 5 assumed one leader per shard was close enough to its writers to ignore network distance; global traffic breaks that assumption.

The cost of a single leader across regions

Chapter 0 established that writes must reach the leader and wait for its fsync before they are acknowledged. If every shard's leader lives in one region, every write from every other region pays that region's round-trip network cost on top of the fsync itself. Light in fiber travels at roughly 200,000 km/s; Tokyo to Virginia is about 10,900 km one way:

Real network paths add routing overhead on top of that physical floor, typically landing observed round trips somewhere in the 150–200ms range for this distance. Every single write from a Tokyo user, under a single-region-leader design, pays that cost before it is even acknowledged — independent of Chapter 5's batching, which helps the fsync itself but does nothing for the network hop to reach it.

Multi-leader replication: accept writes locally, replicate after

Multi-leader replication puts a leader in each region — Virginia, Tokyo, Frankfurt — each independently accepting local writes and durably committing them locally, then asynchronously replicating those writes to the other regions' leaders in the background. A Tokyo user's write commits against the Tokyo leader in roughly this lesson's normal single-region fsync time, not a 150ms cross-ocean round trip:

A 30–150× latency improvement for regional users, at a cost this chapter now derives precisely: two leaders can now each independently accept a write to the same logical record at nearly the same instant, and something has to decide what the record's value is once both writes have replicated everywhere. Single-leader designs never faced this question at all — there was only ever one place a write could originate, so there was never a second opinion to reconcile.

The conflict, made concrete

Picture a specific record this ad network tracks: a campaign's remaining budget, decremented each time it's charged for a click. Two spends happen almost simultaneously in different regions:

Both writes are individually valid, computed from the same starting value, 30ms apart, by two leaders that had no way to know about each other in real time. The system now has two candidate values for the same record and must pick, deterministically, the same way on every replica, or different regions will disagree forever about this campaign's budget.

Last-write-wins, and why clocks make it dangerous

The simplest resolution rule is last-write-wins (LWW): attach a timestamp to every write, and when two writes conflict, keep whichever has the later timestamp, discard the other. It is simple to implement and deterministic — every replica, given the same two timestamped writes, picks the same winner. It is also dangerous, because “later timestamp” depends on clocks that are not perfectly synchronized across regions.

Real distributed systems rely on protocols like NTP to keep clocks close, but “close” is not “identical” — typical NTP-synchronized clock skew across data centers runs on the order of a few milliseconds, and can spike far higher under network issues or misconfigured time sources, occasionally into hundreds of milliseconds or worse. Suppose instead Virginia's clock is running 40ms behind Tokyo's true time. Virginia's 09:00:00.030 write — which genuinely happened second, thirty milliseconds after Tokyo's — gets stamped by its own lagging clock as 08:59:59.990, reading as though it happened before Tokyo's 09:00:00.000 write rather than after it. LWW would then keep the earlier, already-superseded Tokyo write and discard the Virginia write — the one that a human observer, watching a correctly synchronized clock, would say actually happened more recently. The direction matters: a fast clock only ever makes a write's timestamp read later than it should, which cannot invert a true ordering; only a slow clock can make a genuinely later write appear earlier, and that is the specific failure mode worth checking for.

Pricing the mistake

Follow the Tokyo/Virginia example through to its financial consequence — the baseline case first, with perfectly synchronized clocks and no skew at all, just two genuinely concurrent writes to the same record. LWW keeps Virginia's $9,925, the write with the later, correctly recorded timestamp, and discards Tokyo's $50 charge entirely — not merged, not double-counted, simply gone, because LWW does not add two writes together, it picks one and throws the other away, even when every clock involved is perfectly accurate:

Fifty dollars, from one lost write. At this ad network's scale — a small fraction of 1,000,000 writes/sec involve genuinely concurrent cross-region updates to the same record, but even a tiny fraction of a million-per-second stream is a large absolute number of incidents per day, each silently under-tracking real ad spend the finance team believes is being billed accurately. This is not a rare edge case worth ignoring; it is a systematic, compounding error baked into the resolution rule itself, and clock skew, from the section above, only makes it worse: a slow clock on either region's leader can flip which of the two writes LWW keeps, so the $50 lost is not even reliably the earlier or the smaller of the two charges — on a bad day it can just as easily be the fresher, larger one that vanishes instead.

Run the same arithmetic at a conservative estimate of 0.001% of writes hitting a genuine cross-region conflict — one in a hundred thousand:

Forty-three thousand dollars a day is not a rounding error in an ad network's books; it is a reconciliation problem large enough to be noticed eventually, at which point the question becomes not whether to fix it, but how much unrecoverable discrepancy accumulated before anyone looked.

The rest of this chapter is about not waiting to find out — choosing a resolution strategy deliberately, per record type, before the first conflict happens rather than after, starting with why the simplest available rule is also the riskiest one.

A better fix for ordering itself: logical clocks

Before reaching for a merge strategy like a CRDT, it is worth asking whether the ordering problem itself can be fixed, rather than worked around. The root cause above was trusting wall clocks — physical clocks synchronized imperfectly across machines — to order events that happened on different machines. A hybrid logical clock (HLC) combines a physical timestamp with a logical counter that increments whenever a message is sent or received, so that if event A is known to have causally influenced event B (B's leader received a message that included A's timestamp before B happened), the HLC guarantees B's timestamp is strictly later than A's — a guarantee raw wall-clock time cannot make under skew.

HLCs do not solve the Tokyo/Virginia scenario in this chapter directly, because those two writes are genuinely concurrent — neither one caused or was aware of the other, so there is no causal relationship for an HLC to correctly order, and it is honest, not a bug, that a system cannot know which of two truly independent events “really” happened first. What HLCs do fix is the more common and more dangerous case: two writes that actually do have a causal relationship (a read-then-write pattern, replicated across regions) getting mis-ordered purely by clock skew, which is precisely the kind of bug LWW on raw wall-clock time is prone to introduce silently.

How real multi-leader systems actually resolve this

This is not a hypothetical design space — production multi-leader and multi-region systems make an explicit, documented choice here, and it is worth seeing the spread:

None of these choices is universally correct. The right answer depends on whether the data shape tolerates LWW's occasional silent loss (view counts, most analytics), needs a CRDT's guaranteed merge (counters, sets), or needs to pay consensus's latency cost to avoid the conflict outright (financial ledgers, inventory counts where correctness cannot be probabilistic).

A better fix for this specific shape of conflict: CRDTs

The Tokyo/Virginia scenario has a specific shape LWW is badly suited to: both writes were decrements, and the correct resolution is not “pick one,” it is “apply both.” A CRDT (conflict-free replicated data type) is a data structure designed so that merging two concurrent updates always produces a mathematically well-defined, order-independent result — no timestamp comparison, no picking a winner, no clock trust required.

For a decrementing counter, the relevant CRDT is a PN-counter (positive-negative counter): instead of storing one number, each replica keeps its own separate running total of increments and decrements it has personally applied. Merging two replicas is simple, commutative addition — take the larger of each replica's individually-tracked counts, per replica, and sum:

PN-counter merge, Tokyo + Virginia
Tokyo's local ledger:     P=0, N=50     (Tokyo applied one $50 decrement)
Virginia's local ledger:  P=0, N=75     (Virginia applied one $75 decrement)

merged N = max(Tokyo.N, Virginia.N) per replica, summed across replicas = 50 + 75 = 125
final budget = $10,000 − $125 = $9,875 — the correct answer, no clock involved

Both decrements survive the merge, because the CRDT was designed so merging never discards information — it combines it, deterministically, regardless of what order the two replicas learn about each other's writes in, and regardless of any clock reading at all. The cost is real: CRDTs only exist for specific data shapes (counters, sets, certain map structures) with well-defined merge semantics, and an arbitrary record like “this row's JSON blob” generally has no natural CRDT equivalent — for those, LWW or an application-specific merge function remains the pragmatic choice, applied carefully with the clock-skew risk above explicitly acknowledged rather than assumed away.

It is worth being precise about what “merge-friendly” means before reaching for a CRDT everywhere. A counter merges cleanly because addition is commutative and associative — order never matters to the final sum. A set merges cleanly under union for the same reason. A record with fields that depend on each other — a campaign's status field that should only transition forward through a fixed sequence of states, say — does not have an obvious commutative merge, and forcing one risks producing a value nobody ever actually wrote. Reach for a CRDT when the underlying operation is genuinely commutative; reach for LWW, a causal clock, or consensus otherwise.

Concept → realization: LWW resolution, and where it silently fails

python
def resolve_lww(write_a, write_b):
    # write_a and write_b each carry a wall-clock timestamp from their originating leader
    if write_a.ts >= write_b.ts:
        return write_a          # write_b is silently discarded — no error, no log, nothing
    return write_b

def resolve_pn_counter(counter_a, counter_b):
    merged = PNCounter()
    for replica_id in set(counter_a.replicas) | set(counter_b.replicas):
        merged.P[replica_id] = max(counter_a.P.get(replica_id, 0), counter_b.P.get(replica_id, 0))
        merged.N[replica_id] = max(counter_a.N.get(replica_id, 0), counter_b.N.get(replica_id, 0))
    return merged             # every replica's contribution survives — nothing discarded

Line one of resolve_lww is where the entire risk of this chapter lives: a single comparison, no logging of what got thrown away, no signal to any operator that data was lost. Contrast resolve_pn_counter, where the shape of the merge itself makes information loss structurally impossible — there is no branch in that function that discards a replica's contribution, which is precisely why CRDTs are worth the added complexity for the specific data shapes they support.

Sensitivity: how bad can clock skew realistically get

That last row is worth a specific mention: Google's Spanner does not claim clocks are perfectly synchronized, it claims to know the maximum possible error and waits out that uncertainty window before committing a transaction that depends on ordering — a fundamentally different, more expensive, more honest answer than assuming clocks agree and hoping.

What this means for the shard fleet

Multi-leader replication is not the default posture for this lesson's 10 batched shards — it is a targeted tool for the specific records where cross-region write latency genuinely matters and where either the data shape tolerates LWW's risk or a CRDT cleanly fits. Most of this ad network's click events have no meaningful cross-region conflict at all (a click from Tokyo and a click from Virginia are different rows, never contending for the same record), and those shards stay single-leader, exactly as built in Chapters 2 through 5. Multi-leader earns its complexity only where two regions can legitimately race to update the very same piece of state, like the campaign-budget counter this chapter used as its running example.

This is a deliberate, narrow scope for a genuinely complex tool. Reaching for multi-leader replication across an entire schema, for every table, because a handful of records benefit from local write latency, imports this chapter's entire conflict-resolution surface area onto data that never needed it — a click event, uniquely identified and never contended for, gets none of the benefit of multi-leader replication and all of its operational cost. Scope the decision per record type, not per deployment.

Chapter 7: LSM vs B-Tree

Every chapter so far treated “the leader” as a black box that accepts a write, fsyncs it, and stores it somewhere durable. This chapter opens that box. The data structure a storage engine uses to actually organize bytes on disk has its own write cost, entirely separate from the fsync latency Chapter 0 measured — and for a write-heavy stream like this ad network's, that choice is worth as much attention as everything built so far.

It is a genuinely separate axis from every prior chapter. Sharding, partition keys, consistent hashing, and batching all changed how writes are routed and grouped before they ever reach a storage engine; this chapter asks what happens the instant a write actually lands, on one specific leader, and how much extra physical work that landing costs beyond the logical bytes the application asked to persist.

How a B-tree writes a single row

Postgres, like most traditional relational databases, stores table data in a B-tree — a balanced tree of fixed-size pages, typically 8KB each, where each page holds many rows plus pointers to child pages. Finding a row means walking down the tree from the root; writing a row means finding the right leaf page, modifying it in place, and marking that page dirty for the next flush to disk.

The trouble for a write-heavy stream is exactly that “in place.” A stream of 1,000,000 unrelated click events, each with a fresh, effectively random primary key, touches leaf pages scattered unpredictably across the entire tree — there is no reason two consecutive click events land in the same 8KB page. Every one of those scattered pages, once dirtied, eventually has to be written back to disk as a whole 8KB unit, even though the actual new data might be only 200 bytes. Contrast this with a workload where writes are naturally clustered — sequential IDs, say, where consecutive writes land on the same or adjacent pages — and a B-tree's in-place cost shrinks dramatically, because many logical writes share the physical cost of one page rewrite. This ad network's workload has no such clustering.

Deriving a B-tree's write amplification, by hand

Call write amplification the ratio of bytes physically written to disk versus bytes of real, logical data the application asked to write. For a single-row insert touching one leaf page (and, on the occasional page split, a parent index page too — call it 1.3 pages touched on average per write, to account for splits):

Fifty-three bytes physically written to disk for every one byte of logical click data, purely from the cost of rewriting whole pages for small, scattered updates. Under contention, with larger indexes and more frequent page splits, this figure commonly runs higher still — real-world B-tree write amplification for small random writes is frequently cited in the 100–200× range, which is the figure this lesson uses going forward as a representative worst case for this specific workload shape.

How an LSM-tree writes the same row

A log-structured merge-tree (LSM-tree) takes a fundamentally different approach: never modify data in place at all. A write first lands in an in-memory sorted structure called a memtable (and, for durability, an append to the same kind of write-ahead log this lesson has used throughout). Once the memtable fills, its entire contents are written to disk in one sequential pass as an immutable file called an SSTable (sorted string table). No existing file is ever edited — new data always becomes a brand new, sequentially-written file.

At flush time, writing 200 bytes of real data costs close to 200 bytes of physical I/O — no wasted page space, because the write is packed sequentially and tightly rather than slotted into a fixed-size page at a scattered location. The write amplification at this stage alone is close to 1×, the honest floor.

Compaction is not free: LSM's own amplification

The catch is step 3. SSTables are immutable, so an update to an existing key does not modify the old SSTable — it writes a new entry to a new SSTable, leaving the old, now-stale entry sitting in an older file. Left unchecked, reads would have to check every SSTable ever written to find the most recent version of a key, and disk usage would grow forever even for a fixed-size dataset. Compaction periodically merges SSTables together, discarding stale entries and producing fewer, larger, up-to-date files — and every byte a compaction pass merges gets physically rewritten, on top of the byte's original flush-time write.

A common tuning, leveled compaction, organizes SSTables into levels of increasing size — each level roughly 10× larger than the one above it — and a byte written at level 0 gets rewritten roughly once per level it eventually merges down through. For a database with, say, 5 levels before data is considered “settled”:

Ten to thirty times, not one — a real, honest cost, but still meaningfully less than the B-tree's 100–200× figure for this workload's shape. The core tradeoff, stated precisely: a B-tree pays its amplification cost immediately, on every individual write, scattered randomly across disk; an LSM-tree pays a much smaller cost at flush time and defers the rest to background compaction, which can be scheduled, throttled, and run sequentially rather than fighting foreground writes for the same random I/O.

That deferral is the whole trade, stated once more plainly: pay a little now and defer the rest to a process with no client waiting on it, or pay a lot now, on the same critical path every client request is blocked on.

Putting a throughput number on the difference

Translate amplification into achievable throughput against a concrete storage budget. A representative NVMe SSD sustains roughly 500 MB/s of sequential write bandwidth, and, separately, roughly 50,000 random 8KB write IOPS — sequential and random I/O are genuinely different physical operations on flash, and each has its own ceiling.

The B-tree's random writes are IOPS-bound. At 50,000 IOPS of 8KB pages:

The LSM-tree's writes are sequential and bandwidth-bound, not IOPS-bound:

Roughly a 7.5× advantage for the LSM-tree on this exact hardware, for this exact workload shape — a direct, physical consequence of trading random I/O for sequential I/O, not a difference in raw disk speed. The disk itself did not get faster or slower between these two calculations; only the pattern of access to it changed. Convert to this lesson's units, at 200 bytes per click event:

Can a B-tree be tuned to close this gap?

It is fair to ask whether the B-tree figure above is worst-case rather than representative. Postgres offers real mitigations: fillfactor tuning leaves deliberate empty space in each page so in-place updates are less likely to trigger a page split, and HOT updates (heap-only tuples) let certain updates avoid touching index pages at all when the updated columns are not indexed. Both genuinely help, and a well-tuned Postgres instance can meaningfully undercut the 100–200× figure used above.

Neither mitigation changes the fundamental shape of the problem, though: writes to a B-tree with scattered keys are still, physically, random-offset operations, and every one of these tunings is trading some other resource (wasted page space for fillfactor, index-freshness constraints for HOT) to soften, not eliminate, the random-write cost. An LSM-tree does not need this category of tuning at all, because its write path was never structured around in-place modification to begin with — the comparison in this chapter is not “untuned B-tree versus tuned LSM-tree,” it holds even after B-tree-side tuning is applied, just by a smaller margin.

Sensitivity: the range this chapter's conclusion is robust across

Even at the most conservative end of this range — a heavily-tuned B-tree against a compaction-heavy LSM-tree — the LSM-tree still comes out ahead for this specific workload shape. The exact multiplier is worth re-measuring on real hardware and a real access pattern before committing to a production number; the direction of the conclusion is not sensitive to getting these constants slightly wrong.

Concept → realization: what compaction looks like on disk

LSM-tree SSTable layout, mid-compaction
Level 0:  [sstable-041] [sstable-042] [sstable-043]     ← freshest, smallest, most overlap
Level 1:  [sstable-A] [sstable-B] [sstable-C] ...        ← ~10× larger than L0
Level 2:  [sstable-X] [sstable-Y] ...                     ← ~10× larger than L1

# background compaction picks overlapping files and merges them:
merge(sstable-042, sstable-B) → new_sstable-B2
  - discards any key present in BOTH, keeping the newer version
  - writes the merged result sequentially, as one new immutable file
  - the old sstable-042 and sstable-B are deleted only once new_sstable-B2 is durable

Nothing in this process ever seeks to a random offset and rewrites a few bytes in place — every operation, flush or compaction, is a sequential read of existing files and a sequential write of a new one, which is exactly the property that keeps this engine's I/O pattern friendly to how flash storage (and, historically, spinning disks) actually perform best.

Compaction runs as a background process precisely because it can: unlike a B-tree's page dirtying, which happens synchronously as a direct consequence of the foreground write, an LSM-tree's compaction has no hard deadline tied to any individual client request. It can be throttled to consume a fixed I/O budget, deferred during a traffic spike, and caught up during a quiet period — a scheduling flexibility a B-tree's write path simply does not have.

Sizing the memtable and WAL segment, concretely

The memtable is not unbounded — it lives in memory, and its size is a deliberate tuning knob with real consequences at either extreme. Size it too small and the engine flushes constantly, producing many tiny SSTables that immediately need compacting (defeating the sequential-write advantage by turning it into frequent small operations); size it too large and a crash loses more unflushed data (bounded by the WAL, which is why the WAL exists independently of the memtable) and a single flush becomes a large, bursty I/O event.

A typical working figure: a 64MB memtable, holding roughly:

At this lesson's batched-and-sharded rate of 100,000 writes/sec per shard (Chapter 5), one memtable fills roughly every:

A new SSTable flush, roughly every 3.2 seconds, per shard, continuously, for as long as this shard is under peak load — a steady, predictable cadence that a compaction scheduler can plan around, rather than the unpredictable, scattered page-dirtying pattern a B-tree produces under the same load.

Real systems, and where this lesson's storage choice lands

That last row is worth pausing on: Kafka's own storage, sitting in front of the sharded database tier this chapter has been analyzing, is not a B-tree or even an LSM-tree — it is a plain append-only log, the simplest possible write path, with none of either structure's overhead. This is exactly why Chapter 5 could treat the queue as absorbing near-unlimited burst throughput: its storage engine was built for nothing but sequential appends from the start.

What this means for this lesson's 10 shards

Given the choice between an 8-shard fleet on a B-tree engine and the same total capacity on an LSM-tree engine, the LSM-tree's roughly 7.5× per-disk throughput advantage, derived above, directly reduces the shard count needed for a fixed write target — or, held at the same 10 shards Chapter 5 arrived at, buys substantial headroom against future growth without adding a single additional shard. For a workload defined from the outset as scattered-key, high-volume, write-dominated ad-click ingestion, an LSM-backed engine is not a marginal optimization; it is the storage-engine decision that matches the workload this entire lesson has been building toward.

What this costs on the read side

LSM-trees do not get this write advantage for free on every axis. A read for a specific key may need to check the memtable, then potentially several SSTables across several levels, since the most recent version could be in any of them — called read amplification. Real systems mitigate this with bloom filters (a compact, probabilistic structure that can quickly say “this SSTable definitely does not contain this key” for the vast majority of irrelevant files, at the cost of the memory footprint), keeping typical read costs close to a B-tree's despite the underlying structural difference. This lesson's focus is the write path, but a fair comparison has to acknowledge the read-side cost this trade introduces, mitigated rather than eliminated.

For this ad network specifically, that tradeoff lines up well with the actual access pattern: the write path (click ingestion) is the aggressive, real-time-critical side this entire lesson exists to serve, while reads (billing reports, analytics dashboards) are comparatively rare, often batched, and far more tolerant of the small extra latency bloom-filter-assisted lookups add. A workload with the opposite shape — rare writes, constant latency-sensitive point reads — would reasonably make the opposite choice, favoring a B-tree's simpler, single-path read cost over an LSM-tree's superior write throughput.

Chapter 8: Assembling the Write Path

Every layer up to this point has been studied in isolation. Now put them in the order a real click event actually travels through them, at the actual target from Chapter 0 — 1,000,000 writes/sec — and watch what happens when one specific key gets unlucky, live, on the assembled system.

Read this chapter as the payoff for the previous eight. Nothing here is new mechanism — every formula and every number below was already derived, in isolation, in an earlier chapter. What is new is seeing them compound against each other, at the real target, on a system that can actually fail in a specific, watchable way and then recover.

The full pipeline, stage by stage

It matters that the router sits after the queue and before the storage engine, not somewhere else. Kafka does not need to know anything about shards — it just needs to hold events durably until a consumer is ready. The router's only job is deciding, deterministically, which of the shards downstream owns a given key. Everything before the router is undifferentiated traffic; everything after it is one shard's problem, and one shard's problem only.

Sizing the fleet, with real headroom

Chapter 5 landed on a bare-minimum 10 shards: 1,000,000 ÷ 100,000 per-shard batched ceiling. A bare minimum is not what gets provisioned in production, for the same reason Chapter 0 flagged: page at 70% of a measured ceiling, not 100%, because bursts are never perfectly smooth and a fleet running flat-out has no margin for the ordinary noise of real traffic. Apply that 70% target to the batched ceiling:

Fifteen, not ten. At 15 shards, average load per shard is:

Comfortably under the 70% alarm line, with room for the ordinary variance real traffic has that a single back-of-envelope average never captures. This is the same “+2 for redundancy” instinct the companion reads lesson applied to its replica count, expressed here as a percentage margin instead of a fixed number of spare boxes — because unlike a replica, an extra shard does not sit idle waiting for a failure, it quietly absorbs its fair share of load every second.

Routing onto the ring, concretely

Each of the 15 shards gets 150 virtual points on the consistent-hashing ring from Chapter 4:

The raw click-event firehose is keyed by click_id — Chapter 3's Fix 1, chosen specifically because a fresh, globally-unique identifier has no popularity distribution to concentrate. Route 1,000,000 of those a second onto the ring, and the law of large numbers from Chapter 4 does its job: with 150 virtual points smoothing out placement luck, each of the 15 shards lands within a few percent of its fair ¹⁄₁₅ share, right around the 66,667/sec computed above. This part of the system, by design, has nothing left to go wrong.

The second write path this pipeline also carries

Not every write in this system is a click event. This ad network also enforces real-time campaign budgets: every click that carries a bid also decrements that campaign's remaining budget, so a campaign cannot overspend between billing cycles. That decrement is a different logical write — same event, same 1,000,000/sec rate, but this one is keyed by campaign_id, because “how much budget does campaign 44012 have left” is fundamentally a per-campaign question, not a per-click one. It rides the same Kafka buffer, the same router, and lands on the same 15-shard ring — on whichever shard campaign_id happens to hash to.

Chapter 3 already derived this exact key's shape: Zipf-distributed, top campaign at roughly 10% of all traffic, or 100,000 decrements/sec, all landing on whichever single shard that one campaign_id hashes to. That shard was already carrying its fair share of the click-event firehose, about 66,667/sec. Add the hot campaign's decrement stream on top:

This is Chapter 0's wall, rebuilt one more layer up, on one shard out of fifteen — and it is not a design flaw in anything built so far. Every layer did exactly what it was supposed to do: the ring balanced shard territory fairly, batching multiplied the ceiling honestly, the LSM engine absorbed the sequential write cost cheaply. The problem is a single real-world campaign being more popular than the average campaign, landing its entire skewed share on one physical machine because campaign_id, unlike click_id, was never going to be unskewed in the first place.

Reading the alarm, and sizing the fix by hand

The per-shard QPS dashboard from Chapter 3 catches this well before the hard ceiling breaks: 166,667/sec crosses the 70,000/sec alarm line the instant the viral campaign's traffic starts arriving, giving whoever is on call real lead time before the shard's queue starts growing without bound. The fix is Chapter 3's Fix 2, applied to exactly the one key that needs it: salt campaign_id into B buckets, spreading its decrements across B different shards instead of one.

Size B from the arithmetic already on the page, the same way Chapter 3 did. Each shard that absorbs a piece of the hot campaign already carries its own 66,667/sec of background click traffic, leaving:

Zero margin is not a real margin, so round up. At B = 4:

Four buckets, not three, not ten. Three leaves no room for the ordinary bursts real traffic has; ten would spread the hot campaign so thin that four of its buckets land on shards that barely notice it, wasting salting's precision for no benefit. 91.7% is still elevated — those four shards are worth watching more closely than the other eleven — but every one of them is under its hard ceiling, with real headroom, and the backlog that was growing without bound at 166.7% stops growing entirely.

Sensitivity: what if this campaign keeps growing

Ten percent was Chapter 3's derived figure for a 12,000-campaign platform, not a hard ceiling on how popular a single campaign can get — a genuinely viral launch can outrun it. Recompute the fix for a campaign that reaches 20% of all traffic, twice the modeled figure:

The bucket count scales with the hot key's share, not with the fleet size — doubling one campaign's popularity roughly doubles the salt buckets it needs, independent of whether the underlying fleet has 15 shards or 1,500. This is the same detection-then-response loop from Chapter 3, running continuously rather than as a one-time calculation: watch per-shard QPS, and when a key's bucket count stops keeping its shards under the alarm line, widen it.

The four numbers this assembled system watches

Every earlier chapter flagged its own alarm signal. Stacked together, they are the honest monitoring surface for the whole pipeline — and, as with the companion reads lesson, notice what is deliberately absent: no single fleet-wide average appears anywhere in this list, because Chapter 3 already showed exactly how an average hides the one number that matters.

Concept → realization: the full write path, end to end

python
HOT_KEYS = {"campaign:44012": 4}          # campaign_id → salt buckets, tuned from live per-shard QPS
RING = ConsistentHashRing(virtual_nodes=150)  # Chapter 4 — 15 shards on the ring
BATCH_WRITERS = {shard_id: BatchWriter(max_batch=100, max_wait_ms=10) for shard_id in RING.shards()}

def route_click(event):
    # the raw firehose — keyed by click_id, never salted, never skewed
    shard_id = RING.shard_for(event.click_id)
    BATCH_WRITERS[shard_id].add(event)              # Ch5 batching, one fsync per 100

def route_budget_decrement(event):
    # the campaign-keyed stream — the only one that ever needs salting
    buckets = HOT_KEYS.get(f"campaign:{event.campaign_id}", 1)
    salt = random.randint(0, buckets - 1) if buckets > 1 else 0
    shard_id = RING.shard_for(f"{event.campaign_id}#{salt}")
    BATCH_WRITERS[shard_id].add_decrement(event.campaign_id, event.charge)

def on_shard_qps(shard_id, qps):                    # called continuously from the per-shard dashboard
    if qps > 0.70 * PER_SHARD_CEILING:
        page_oncall(shard_id, qps)                    # Ch0's 70% alarm threshold, applied live

Two write paths, one ring, one batching layer, one alarm rule. Everything Chapters 0 through 7 built shows up here as a small piece of a bigger whole: the ring from Chapter 4, the batching from Chapter 5, the salting from Chapter 3, the alarm threshold from Chapter 0. None of it is new; all of it is load-bearing.

If a region needs to write locally: folding in Chapter 6

This ad network serves clients globally, so the budget-decrement stream specifically — the one record every region can legitimately race to update — is exactly the narrow case Chapter 6 scoped multi-leader replication for. Each region gets its own leader for that specific counter, accepting decrements locally in a few milliseconds instead of paying a 150ms+ cross-ocean round trip, and merges use the PN-counter from Chapter 6 rather than last-write-wins: every region's decrements survive the merge, regardless of clock skew, regardless of which region's write happened to arrive first at any other region. The raw click-event firehose, keyed by click_id, never needs this — a click from Tokyo and a click from Virginia are different rows that never contend for the same record, so those 15 shards stay single-leader, exactly as built in Chapters 2 through 5. Multi-leader earns its complexity on the one write path that actually has a conflict to resolve, and nowhere else.

What this costs, against the alternative from Chapters 1 and 5

Price the fully assembled system the same honest way every earlier chapter priced its own piece. Fifteen shards, each still the same $620/mo Small-tier box from Chapter 1 — batching changed what that box can do, not what it costs:

†Ballpark figure for a managed streaming cluster at this throughput, rounded to illustrate scale — exact pricing varies by provider and retention window.

Compare against the $920,000/month figure Chapters 1 and 5 costed for sharding alone, unbatched — a ≈60× reduction, from the same 1,000 shards down to 15, purchased entirely by batching, correct partition-key choices, and salting exactly the one key that needed it rather than over-provisioning the whole fleet against a worst case that only ever touches a handful of keys. It is also, worth noting directly, cheaper in total than a single Chapter 1 X-Large-tier vertical box ($8,100/month) that could not have reached the target at all — the assembled system is not just dramatically cheaper than the naive horizontal answer, it is cheaper than a single box that was never going to work in the first place.

What we built, and what it is worth

None of these layers is optional past a certain scale, and none of them is free below it — every one adds an operational surface (a ring to maintain, a salt table to keep current, a compaction scheduler to tune) in exchange for write throughput the single leader in Chapter 0 could never have absorbed. A small analytics pipeline logging a few hundred events a second does not need any of this — the single Small-tier box from Chapter 1, entirely unbatched, comfortably clears that traffic, and every layer this lesson built would be pure operational overhead with no write volume to justify it. The value of deriving every number by hand, rather than being told “shard it, batch it, use an LSM store,” is that the arithmetic tells you exactly when that stops being true for a specific system, instead of leaving it as a guess.

→ Scaling Reads — the read-side twin of this lesson: replicas, caching, and the funnel that keeps a database tier small on the other side of the same box from Chapter 0
→ Partitioning & Storage — a deeper look at range and hash partitioning across a wider set of real systems
→ Database Replication — the fuller mechanics of leader-follower and multi-leader replication this lesson applied narrowly in Chapter 6
→ Storage & Retrieval — B-trees, LSM-trees, and the storage-engine internals Chapter 7 built on