The question this answers
Every shard has a good p99. Why is the p50 of my fan-out request terrible?
A scatter-gather that waits for all responses has a latency equal to the maximum over its components, so its distribution is the component distribution raised to the power of the fan-out width. There is no configuration that avoids this — only reducing the width, truncating the wait with a deadline, or accepting partial results.
Everything below is bought to hold this sentence. "Strongly consistent" with no scope attached is a slogan, not a guarantee — read what it actually covers, and what it explicitly does not.
The aggregator knows which responses have arrived and how long it has waited. It does not know whether an outstanding shard is about to answer or has 400ms of GC left to run, so it cannot distinguish "nearly done" from "hopeless". Every decision it makes — wait, hedge, or answer partially — is made without knowing which. That is why the deadline, not a prediction, has to be what ends the wait.
A node knows its own state and the messages that arrived. Everything else is inference from evidence that was already stale. "B has not replied in five seconds" is knowledge; "B is down" is a decision — and usually the bug.
The arithmetic, exactly
Let each component independently be "slow" with probability q. The aggregate is slow if any component is slow, with probability 1 − (1 − q)^n. Put in the numbers everyone quotes: q = 0.01, n = 100. Then 0.99^100 = 0.366, so the aggregate is slow 63.4% of the time. A one-in-a-hundred event at the component becomes a two-in-three event at the system.
Sharpen it into percentile terms, which is more useful than the probability. The aggregate is the maximum of n draws, so F_max(x) = F(x)^n. The aggregate median solves F(x)^100 = 0.5, giving F(x) = 0.5^(1/100) = 0.9931. The median of a 100-way fan-out is the component’s 99.31st percentile. The aggregate p99 solves F(x)^100 = 0.99, giving F(x) = 0.99^(1/100) = 0.99990, the component’s 99.99th percentile.
This is the sentence worth carrying away: at a fan-out of 100, your users experience your component p99.3 as a typical request and your component p99.99 as their bad day. Optimising component median latency does essentially nothing for them; only the far tail matters. Teams routinely spend quarters on the median and are surprised that aggregate latency does not move — the arithmetic said it would not.
Fan-out width enters as an exponent, so it is by far the strongest lever. Halving n from 100 to 50 moves the aggregate median from component p99.31 to p98.62 — a much bigger improvement than any plausible reduction in component latency. Reducing width beats optimising components, and it usually is not even close.
| Fan-out n | P(at least one slow), q = 1% | Aggregate median = | Aggregate p99 = |
|---|---|---|---|
| n = 1assumption | 1.0% | component p50 | component p99 |
| n = 10assumption | 9.6% | component p93.3 | component p99.90 |
| n = 50assumption | 39.5% | component p98.62 | component p99.980 |
| n = 100assumption | 63.4% | component p99.31 | component p99.990 |
| n = 1000assumption | 99.996% | component p99.931 | component p99.9990 |
The independence assumption cuts both ways
All of the above assumes components are slow independently. They are not. They share racks, top-of-rack switches, storage backends, a control plane, a configuration source, and often a clock-driven event like a cache TTL or a scheduled compaction.
Correlation makes the "at least one" probability lower than the independent calculation — if shards tend to be slow together, then when none is slow, none is slow. So 63.4% is a pessimistic estimate for how often you get a slow aggregate. But correlation makes the bad case much worse: when the shared cause fires, all 100 shards are slow at once, and no amount of hedging, partial results or replica diversity helps, because there is nowhere unaffected to go.
The practical consequence is that hedging and shard-level redundancy work on the *independent* portion of slowness and do nothing for the correlated portion. Measuring which regime you are in is therefore the first diagnostic step, and it is easy: plot the number of slow shards per request. Independent slowness gives a distribution concentrated near zero with a thin tail; correlated slowness gives a bimodal shape — almost none, or almost all. The two demand completely different fixes, and the aggregate latency graph looks similar for both.
slow shards INDEPENDENT CORRELATED 0 36.6% 71.2% 1 37.0% 1.1% 2 18.5% 0.4% 3 6.1% 0.3% 4-9 1.8% 0.6% 10+ 0.0% 26.4% <- shared cause fires hedging helps: yes no (nowhere healthy to hedge to) partial results: yes no (too much of the answer is missing) the fix: width, hedging find the shared cause
The four things that actually work
Reduce the width. It is the exponent, so it dominates everything else. Query fewer shards by routing on a key, adding a coarse index, or pre-aggregating so one call replaces fifty. A design that turns a 100-way scatter into a 5-way one has done more for tail latency than any amount of per-shard tuning could.
Do not wait for everyone. Return at the deadline with what you have and mark the response as partial. For search, recommendations, feeds and analytics this is almost always the right call: 95 of 100 shards at 80ms beats 100 of 100 at 900ms, and the user cannot tell what is missing. This requires the API to express partiality honestly — a partial result silently presented as complete is a correctness bug, not a latency optimisation. That is exactly the concern API Design owns as partial-failure.
Hedge the stragglers. Once a shard passes the threshold, ask its replica. This works precisely on the independent portion of slowness, and it composes well with a deadline. See Send a Second Request After p95 and Take Whichever Answers First.
Make the units smaller than the servers. Instead of one partition per server, cut the data into many more partitions than servers — say 20 per server. A slow server then holds 20 small pieces rather than one large one, and those pieces can be migrated away or served by replicas at fine granularity. Load imbalance and straggler impact both shrink, because the scheduler has smaller units to move. *The Tail at Scale* calls this micro-partitioning, and it is the same idea as virtual nodes in Virtual Nodes: Many Positions per Machine, and Why It Is Not Optional, applied to latency rather than to rebalancing.
1async function scatterGather(shards: Shard[], q: Query, budgetMs: number) {2 const results: Row[] = []3 const missing: string[] = []4 const ctl = new AbortController()5 6 const calls = shards.map(async (s) => {7 try { results.push(...await s.query(q, ctl.signal)) }8 catch { missing.push(s.id) }9 })10 11 // Wait for everyone OR the budget, whichever comes first. Waiting for12 // everyone means waiting for the maximum, which is the whole problem.13 await Promise.race([Promise.allSettled(calls), sleep(budgetMs)])14 ctl.abort() // stragglers are cancelled, not left running15 16 const answered = shards.length - missing.length17 return {18 rows: results,19 // Partiality must be visible. A partial answer presented as complete is a20 // correctness bug wearing a latency optimisation's clothes.21 complete: missing.length === 0,22 coverage: answered / shards.length,23 missingShards: missing,24 }25}Key points
- A fan-out that waits for all components has the latency of the maximum, so its distribution is the component distribution raised to the power n.
- With n = 100 and a 1% chance of a slow component, 63.4% of requests contain at least one slow component.
- The aggregate median equals the component’s 99.31st percentile at n = 100 — optimising the component median does nothing for the user.
- Width is the exponent and therefore the strongest lever; halving the fan-out beats almost any component-level improvement.
- Correlated slowness makes "at least one slow" rarer but far worse when it happens, and neither hedging nor partial results help in that regime.
The chain, answered
Every field here is required, which is why no lesson in this domain can recommend a design without naming what an operator sees when it fails, what survives the partition, what repairs it afterwards, and the simpler thing to consider first.
- • A request is scattered to n components, each with its own latency distribution.
- • The aggregator waits for responses; if it waits for all, its latency is the maximum of n draws.
- • The probability that the maximum exceeds a threshold is
1 − (1 − q)^n, which rises steeply with n. - • Introducing a deadline truncates the wait: the aggregator answers with the responses it has and marks the rest missing.
- • Hedging replaces individual straggler draws with the minimum of two draws, reducing the effective per-component tail on the independent portion.
- • One component is slow for a reason unrelated to the request — GC, compaction, a cold cache, a noisy neighbour.
- • A shared cause makes many components slow simultaneously, so redundancy has nowhere healthy to route.
- • The aggregator waits for all components with no deadline, so a single straggler sets the response time.
- • Partial results are returned without being labelled, so callers treat an incomplete answer as complete.
- • Cancellation of stragglers is missing, so abandoned component work continues to consume capacity.
- • Aggregate p50 far worse than component p50 while every component dashboard looks healthy. Each shard owner correctly reports good numbers and the user experience is poor — the classic and most confusing signature.
- • Latency dominated by a rotating culprit: every request is slowed by a different shard, so no single shard looks bad over any window. Only per-request straggler attribution exposes it.
- • Bimodal aggregate latency — fast most of the time, then a cluster of very slow requests. Slow-shard-count is bimodal too, indicating a shared cause rather than independent variance.
- • Silent incompleteness: results are missing shards and no field says so. Users see inconsistent result counts between identical queries and nothing is logged as an error.
- • Scatter-gather needs no agreement, which is why it scales — but it inherits the worst component’s timing, which is the price of that freedom.
- • Answering at a deadline requires a decision about completeness that only the aggregator can make, and that decision must be surfaced to the caller rather than buried.
- • Micro-partitioning needs a scheduler that can move small units between servers, which is coordination in the control plane rather than on the request path — the right place for it.
- • With a deadline and partial results, the system degrades in *coverage* rather than in latency, which is usually the better degradation for search and analytics.
- • Without a deadline, one straggler degrades every request that touches it, and the blast radius of one slow shard is the whole fan-out.
- • Correlated slowness degrades everything at once regardless of design, so the honest response there is capacity or removing the shared cause, not a latency technique.
- • Detect: measure aggregate latency and per-shard latency together, and attribute each slow request to the shard that caused it. Straggler attribution is the missing metric in most fan-out systems.
- • Contain: put a deadline on the gather and return partial results, so one slow shard cannot set the response time for everyone.
- • Recover: rebalance or evict a persistently slow shard; with micro-partitioning this is a small, cheap move rather than a large one.
- • Reconcile: track coverage per response so downstream consumers and offline analyses know which results were computed on incomplete data.
- • Verify: inject latency into a single shard and confirm aggregate p99 barely moves. If it moves a lot, the gather is still waiting for the maximum.
- • Slow-shard count per request, as a distribution — the single diagnostic that distinguishes independent from correlated slowness.
- • Straggler attribution: which component was last to respond, per request, aggregated over time. Reveals rotating culprits that per-shard dashboards cannot.
- • Coverage (responses received divided by fan-out width) on every partial answer, exported alongside latency.
- • Component p99 and p99.9 rather than component p50, since those are the percentiles the aggregate actually samples from.
- • Understanding this is mandatory for any scatter-gather: search, distributed queries, feed assembly, multi-shard reads, and any request that touches many services.
- • Especially valuable when deciding where to spend optimisation effort, because it says plainly that component-median work is wasted and width and tail work is not.
- • Narrow fan-outs of two or three, where the amplification is small and the analysis is overhead.
- • Systems where partial results are unacceptable — a financial total, a correctness-critical aggregate — because then the deadline lever is unavailable and only width and tail reduction remain.
- • When the conclusion is misread as "add more replicas": replicas help the independent portion only, and correlated slowness needs the shared cause removed.
- • Route to fewer shards using a partition key or a coarse index, eliminating the fan-out rather than tolerating it.
- • Pre-aggregate into a materialised view so the read touches one place. Trades freshness and write cost for the removal of the exponent — see Materialized Views: A Read Model That Lags.
- • Two-phase fan-out: query a cheap summary layer across all shards, then fetch details from the few that matter.
- • Accept the tail where the operation is rare or offline — a nightly report can wait for the maximum, and none of this machinery is worth it there.
Fan-out tail latency: the arithmetic, exactly
| Partitions queried | P(all within p99) | User median | Effective experience |
|---|---|---|---|
| 1 | 99% | 10 ms | the p99 is the p99 |
| 10 | 90% | 57 ms | one request in 10 hits a slow shard |
| 50 | 61% | 130 ms | one request in 3 hits a slow shard |
| 100 | 37% | 176 ms | most requests hit a slow shard |
| 200 | 13% | 232 ms | most requests hit a slow shard |
What people believe, and what is true
Our shards all have a good p99, so the system is fast.
At a fan-out of 100 the aggregate median *is* roughly the component p99.3. A good component p99 is the input to your typical user experience, not a margin of safety.
Optimising median component latency will fix aggregate latency.
The aggregate samples the far tail of the component distribution. Median work is nearly invisible to it; tail work and width reduction are not.
Add replicas and hedge, and the fan-out tail goes away.
That treats the independent portion. When slowness is correlated across shards there is no healthy replica to hedge to, and the technique adds load without helping.
The 63% number is a solid prediction.
It is the independent-case upper bound. Correlation lowers the frequency and raises the severity. Plot slow-shard count per request to see which regime you are actually in.
Go deeper
Only the levels this lesson can honestly fill — a missing level is a claim nobody had.
Overview
Waiting for 100 components means waiting for the slowest of 100. Even if each is rarely slow, "at least one is slow" is the common case — so your users feel the component tail, not the component average.
Practical
Put a deadline on the gather, return partial results with explicit coverage, cancel the stragglers, and measure slow-shard count per request plus straggler attribution. Then attack width first — fewer shards per request beats faster shards.
Advanced
Work with the order statistic directly: the aggregate is F(x)^n, so an aggregate percentile p maps to the component percentile p^(1/n). That inversion tells you exactly which part of the component distribution to invest in, and it says median work is wasted at any meaningful width. Then check the independence assumption empirically via slow-shard count, because the entire calculation is a different problem in the correlated regime — and micro-partitioning helps in both, by making the unit of imbalance small enough for the scheduler to move.
Apply it
- 💬 A request fans out to 100 shards, each slow 1% of the time. How often is the request slow, and what assumption did you use?
- 💬 Your shards report excellent p99s and your users report bad typical latency. Reconcile those two facts.
- 💬 How would you tell whether shard slowness is independent or correlated, and why does the answer change your fix?