The question this answers
If every shard is fast 99% of the time, how often is a query across all of them fast — and what do I do about the answer?
A search query broadcast to 100 index shards, each returning its top 20 matches, merged into a single ranked list of 20.
The gather buffer holding one partial result per shard, and the merge state that produces the final ranking. The shards themselves share nothing with each other — that is the point of the partition — so all the coordination is at the gather.
The merged result is exactly what a single sorted pass over all shards' contributions would produce, and every shard is represented exactly once — by its results, by an explicit failure, or by an explicit "omitted, deadline exceeded" marker that the caller can see.
Synchronization exists to preserve this sentence. If a schedule can make it false, the code is wrong no matter which primitive it uses.
Every shard fast, the query slow
Scatter/gather is the natural read pattern for partitioned data: the query cannot be answered by one shard, so it goes to all of them, and the coordinator merges. Structurally it is Fan-Out / Fan-In: One Request Becomes N with the branches being homogeneous rather than heterogeneous — the same query, N times, over disjoint data. That homogeneity is why the tail-latency effect is so sharp: you are not waiting for one known-slow dependency, you are waiting for whichever shard happened to be unlucky this time, and with enough shards *some* shard is always unlucky.
The timeline makes it concrete. Ninety-nine shards answer in 8ms. One shard is behind a garbage collection pause, or a cold page cache, or a co-tenant's backup job, and answers in 140ms. The query took 140ms. The coordinator was idle for 132ms of it, the ninety-nine fast results sat in a buffer, and every dashboard for every shard shows a healthy p50 of 8ms.
This is the reason a service can have excellent per-node latency metrics and terrible user-facing latency, and why "which shard is slow" is the wrong question — it is a different shard every time. The right question is what the coordinator does when one shard is late, and that is a design decision you must make explicitly, because the default (wait for everyone) is the worst option available.
The arithmetic that surprises everyone
Treat "slower than this shard's p99" as an independent event with probability 0.01. The probability that *no* shard is slow is 0.99^N, so the probability that the whole query is slow is 1 - 0.99^N. At N=10 that is 9.6%. At N=100 it is 63.4%. Your one-in-a-hundred event has become the common case, and no individual shard did anything wrong.
Read the table the other way and it is even more useful. For the *query* to have a p99 equal to a single shard's p99, each shard would need that latency at its p99.9 (N=10) or p99.99 (N=100). Scatter/gather does not just expose the tail — it demands a tail an order of magnitude tighter than the one you are measuring, per additional order of magnitude of fan-out width. That is why large scatter/gather systems invest so heavily in tail control at the node level: pause-free allocation, request hedging, admission control, and keeping N as small as the data layout allows.
Note also the assumption doing work here: independence. Shard slowness caused by a shared cause — a network partition, a coordinated GC, a hot key routed to several replicas, a noisy neighbour on the same host — is correlated, and correlated failures make the real numbers worse than the table, not better. The table is the optimistic case.
- p99 of the query is not p99 of a shard. It is roughly the p(99^(1/N)) of a shard, which gets extreme fast.
- Halving N is a bigger tail improvement than most per-shard optimizations, and it is a data-layout decision.
- If you cannot measure a shard's p99.9, you cannot reason about a 10-way gather's p99.
assume each shard exceeds its own p99 latency 1% of the time,
independently. the gather waits for the slowest.
N shards P(at least one slow) per-shard percentile needed
for the QUERY to hit p99
-------- -------------------- --------------------------
1 1.0 % p99
5 4.9 % p99.8
10 9.6 % p99.9
20 18.2 % p99.95
50 39.5 % p99.98
100 63.4 % p99.99
read it twice:
left -> more shards means the tail event is no longer rare.
right -> more shards means you must control a tail you are
probably not even measuring.
and independence is the OPTIMISTIC assumption. correlated slowness
(shared host, shared network, coordinated GC) is worse than this.Four ways to stop waiting for the slowest
The coordinator has exactly one lever: what it does at the moment the results are *almost* all in. Waiting is the default and the worst. The alternatives all trade something real, and choosing among them is a product decision as much as an engineering one — "is a search result that is missing one shard's contributions acceptable?" is not a question the coordinator can answer on its own.
Hedging deserves a specific note because it is the least intuitive and often the most effective: after waiting a short interval — commonly around the p95 of a shard query — send a duplicate request for the outstanding shards to a different replica and take whichever answers first. This costs a small percentage of extra load (only the requests that were already slow get duplicated) and can remove most of the tail, because the causes of tail latency are usually *local and transient* rather than properties of the query. It requires idempotent reads and cancellation of the loser, and it is strictly worse than useless if the shard is slow because it is overloaded — hedging into an overloaded replica adds load to the thing that is already failing.
The structural fix underneath all four is to reduce N. Routing a query to the one shard that can answer it — because the partition key is in the query — takes the tail arithmetic off the table entirely. Most scatter/gather in practice exists because the access pattern and the partition key disagree, which makes it a schema problem wearing a concurrency costume (Partitioning and Sharding in Database Engineering is where that conversation belongs).
- Hedging duplicates only the requests that are already slow, so its cost is proportional to your tail, not your traffic.
- Hedging into an overloaded replica makes things worse; gate it on a health signal and a cap.
- A partial result must be visible in the response contract. Silently returning 97 shards' worth of data as if it were complete is a correctness bug, not a degradation.
| Strategy | What it gives up | Extra load | Requires | Good when |
|---|---|---|---|---|
| Wait for all | Nothing — but inherits every shard's tail | None | Nothing | N is small and completeness is mandatory |
| Deadline + partial results | Completeness, visibly | None | A response shape that can say "97 of 100 shards" | Ranked or approximate results (search, analytics) |
| Hedged request after p95 | Little; some duplicated work | ~5% of reads | Idempotent reads, replicas, real cancellation | Tail is transient and local (GC, cold cache) |
| Speculative full duplication | Nothing, but doubles cost | 100% | Replicas with spare capacity | Latency matters far more than cost |
| Reduce N by routing on the key | Query flexibility | Negative — far less | Partition key present in the query | Always worth attempting first |
Key points
- A gather completes at the speed of its slowest branch, so the query inherits the tail of every shard it touched.
- With independent 1% tail events, P(slow query) is 1 - 0.99^N: 9.6% at 10 shards, 63.4% at 100.
- For the query to hit p99, each shard must hit that latency at roughly p99.9 (N=10) or p99.99 (N=100) — a tail most teams do not measure.
- The coordinator's only lever is what it does when results are almost all in: wait, deadline with partial results, hedge, or duplicate.
- Reducing N by routing on the partition key beats every coordinator-side mitigation, and is usually a schema decision rather than a concurrency one.
The loop, answered
Every field is required, which is why no lesson here can recommend concurrency without naming the interleaving that breaks it, the complexity it adds, and the simpler thing to consider first.
- • The coordinator maps the query to the set of shards that could contain matches — all of them, unless the partition key is present in the query.
- • It scatters the query concurrently, each with a per-shard deadline derived from the request budget, and records which shards were asked.
- • Each shard executes independently over disjoint data and returns a partial result — commonly its own top-k, which is what makes the merge cheap.
- • The coordinator gathers into per-shard slots (no shared mutable accumulator), applies its completion policy, and merges: a k-way merge over the partials, re-ranked globally.
- • The response carries provenance: how many shards contributed, and which were omitted, so the caller can distinguish "no results" from "we did not look everywhere".
- • Normal: 100 shards respond in arbitrary order into disjoint slots; the merge runs once, after the last one. Every ordering is equivalent because the slots do not overlap.
- • The tail case: 99 shards respond within 11ms, one responds at 140ms, and the coordinator holds the request — and its memory, its connection and its share of the pool — for 129ms of pure waiting.
- • Deadline case: the coordinator responds at 50ms with 97 shards; shard 98 responds at 140ms and writes into a gather buffer whose request has already ended. Without an explicit guard that is a write to freed or reused state (Orphaned Tasks).
- • Hedging: the coordinator sends a hedge to replica B at 20ms; A answers at 25ms, B at 30ms; the coordinator takes A and must cancel and discard B, or it double-counts shard 100 in the merge.
- • Correlated slowness: a network blip makes twelve shards slow simultaneously, the independence assumption collapses, and hedging duplicates twelve requests into a network that is already the problem.
- • Guarantees a complete answer only under the wait-for-all policy, and only if no shard fails — which is why availability falls as N grows.
- • Guarantees each shard is queried at most once per attempt, and (with disjoint slots) that no shard's result overwrites another's.
- • A deadline guarantees when the coordinator responds. It does NOT guarantee the outstanding shard queries stopped, released their connections, or stopped consuming shard CPU.
- • Hedging guarantees a faster *typical* tail only if the slow cause is local and transient. It guarantees nothing when the shard is slow because it is saturated.
- • Top-k-per-shard merging guarantees a globally correct top-k only when the ranking is decomposable that way — some aggregations (exact distinct counts, global medians) are not, and per-shard partials cannot be merged into an exact answer.
- • The coordinator holds one request's worth of resources for the duration of the slowest shard — at high concurrency that is a large amount of memory and connections tied up doing nothing.
- • Each shard is queried by every coordinator, so shard load scales with total query rate, not with per-coordinator rate; a shard is the shared resource for the entire fleet.
- • The merge itself is CPU work on the coordinator and becomes a bottleneck when N and k are large — a 100-way merge per query is not free.
- • Hedging adds contention exactly where the tail already is, so it must be capped (a fixed small fraction of requests) and disabled under load shedding.
- • Tail amplification: the query p99 is far worse than any shard's p99 and no single shard looks unhealthy.
- • Availability decay: with wait-for-all, a single shard failure fails the entire query, so aggregate availability falls geometrically in N.
- • Silent incompleteness: partial results returned as if complete, so "no matches" and "we did not reach the shard holding the match" are indistinguishable to the caller.
- • Hedge storms: hedging enabled globally during a real overload multiplies load on an already-saturated tier.
- • Duplicate contributions when a hedge and its original both return and the merge does not deduplicate by shard.
- • Coordinator memory exhaustion under load, because every in-flight query holds N partial results while waiting for one straggler.
- • Search and analytics over partitioned data, where the answer genuinely requires every partition and per-shard top-k makes the merge cheap.
- • Embarrassingly parallel read work where per-shard cost is meaningful, so the concurrency converts a large sum into a manageable maximum.
- • Systems with replicas and idempotent reads, where hedging is available as a tail-control mechanism.
- • Approximate or ranked results, where a deadline-bounded partial answer is genuinely acceptable and can be labelled as such.
- • When the partition key is available and the query could have gone to one shard. Then the entire tail problem is self-inflicted.
- • When N is large and completeness is mandatory — the availability and tail arithmetic are both against you and there is no coordinator-side fix.
- • When per-shard work is tiny: the scatter, the network round trips and the merge dominate, and a single larger shard would be faster.
- • When the aggregation is not decomposable, so shards cannot return small partials and the coordinator gathers enormous intermediate results.
- • When shard slowness is correlated, which invalidates both the arithmetic and the hedging strategy.
- • Query p99 next to per-shard p99. The gap between them is the tail-amplification cost, and it is the number that justifies any of this work.
- • Straggler attribution: for each query, which shard was last, and how much of the total wait it caused. If it is a different shard every time, it is a tail problem, not a hot-shard problem.
- • Fraction of queries that hit the deadline and returned partial results — the honest measure of how often completeness is being silently traded away.
- • Hedge rate and hedge win rate: hedges sent as a fraction of requests, and how often the hedge beat the original. A low win rate means you are adding load for nothing.
- • Coordinator in-flight memory and connection occupancy, which is proportional to concurrency times N times partial size and is the usual scaling wall.
- • The response shape must express partiality — shards queried, shards answered, shards omitted — and every consumer must handle it, or the degradation becomes a silent correctness bug.
- • Hedging adds replica selection, deduplication by shard, cancellation of the loser, a cap, and a kill switch for overload conditions.
- • Per-shard deadlines must be derived from a request budget and propagated, which requires deadline plumbing through every layer (Deadlines vs Timeouts).
- • The merge needs its own correctness argument: when top-k-per-shard composes into a global top-k, and when it does not.
- • Observability must be per-shard-per-query to attribute stragglers at all, and that is a high-cardinality telemetry problem in its own right.
- • Route by partition key so the query touches one shard. Removes the tail arithmetic, the merge and the availability decay simultaneously — always evaluate this first.
- • Precompute the aggregate: a materialized view or a rollup answers from one place, trading freshness for a single-shard read.
- • Two-phase: query a small cheap index to identify the few shards worth asking, then scatter to those. Cuts N by an order of magnitude for many workloads.
- • Cache the query result, when repeat queries are common — a cache hit has no tail at all (Caching Architecture in Architecture).
- • Fewer, larger shards, when per-shard capacity allows. N is a variable you control, and it is in the exponent.
One request, N downstream calls
Bounding concurrency with permits
| permits | goodput | mean latency | timeouts | failed of 10K |
|---|---|---|---|---|
| 1 | 25/s | 43 ms | 0.00% | 0 |
| 5 | 125/s | 43 ms | 0.00% | 0 |
| 10 | 250/s | 43 ms | 0.00% | 0 |
| 25 | 625/s | 43 ms | 0.00% | 0 |
| 50 | 1000/s | 53 ms | 0.00% | 0 |
| 100 | 1000/s | 103 ms | 0.00% | 0 |
| 200 | 1000/s | 203 ms | 0.00% | 0 |
| 350 | 1000/s | 353 ms | 0.00% | 0 |
| 500 | 0/s | 503 ms | 100.0% | 10K |
The deadline expired. What happened to the work?
try:
result = await wait_for(call(req), 300ms) # only the *wait* is bounded
except Timeout:
return 504 # call() is still running, on a worker, right nowWhat people believe, and what is true
All the shards are fast, so the query is fast.
The query is as fast as the slowest shard on that attempt. With 100 shards and a 1% tail, some shard is slow on roughly two out of three queries.
p99 of the query should be about p99 of a shard.
It is closer to the p99.9 or p99.99 of a shard, depending on N. Fan-out width shifts which percentile of the component distribution you are actually exposed to.
Hedging is wasteful because it duplicates requests.
It duplicates only the requests that already passed the p95 mark, so the extra load is a few percent. The waste worth worrying about is hedging into a saturated replica.
Returning results from 97 of 100 shards is a graceful degradation.
Only if the caller can see it. Undeclared partial results make "not found" and "not looked for" indistinguishable, which is a correctness bug wearing a degradation costume.