After sharding, p99 read latency got worse and every shard is busy — how do you confirm broadcast queries are the cause?

## The symptom and the wrong instinct The classic report is: "we sharded to scale, and now the same queries are slower and every shard is at the same CPU." The instinct is to add a shard. That is the one action guaranteed not to help, because if the queries broadcast, each new shard adds a participant to every request rather than taking a slice of the traffic. ## Step 1: confirm with explain, per query shape Routing is a per-query property, so diagnose per shape. Through `mongos`: ```js db.orders.find({ status: "OPEN" }).explain("executionStats") ``` What to read: - **Top-level winning-plan stage.** `SINGLE_SHARD` means one shard was contacted. `SHARD_MERGE` means several were and the router merged their output. - **The `shards` array.** It names the participants. If it lists every shard in the cluster, the query is a full broadcast. - **Per-shard `executionStats`.** For each shard you see `nReturned`, `keysExamined`, `docsExamined` and its plan. The tell of a wasteful broadcast is many shards with `nReturned: 0` that nonetheless examined keys or, worse, ran a `COLLSCAN`. - **`SHARDING_FILTER`.** Shard plans include this stage; it discards orphaned documents that a shard still holds from an interrupted migration, so the router never returns them twice. Do this for the top handful of shapes by call volume. Aggregations get the same treatment — the sharded explain shows which shards ran the pipeline and how it was split. ## Step 2: quantify the fan-out Two numbers make the case to anyone: 1. **Shards touched per query.** Weight each hot shape by its call rate. If 80% of calls touch all 20 shards, the cluster is doing roughly 20× the logical work of a single node for those calls. 2. **Per-shard request rate.** On a targeted workload each shard sees about `total_qps / N`. On a broadcast workload each shard sees `total_qps`. Comparing the observed per-shard operation counters against total application throughput settles the question without any explain at all. ## Step 3: understand why p99 in particular degrades Fan-out queries have a structural tail problem. The merge cannot produce its first result until every participating shard has answered, so each request's latency is the **maximum** of N shard latencies, not the mean. If any single shard is momentarily slow — a checkpoint, a competing scan, an election, a noisy neighbour — every in-flight broadcast query inherits that latency. With N shards you draw N samples per request, so your p99 comes from a distribution far to the right of any individual shard's p99. This is why the median can look acceptable while p99 falls apart, and why the effect worsens as you add shards. Secondary amplification: each broadcast consumes a connection and a worker thread on every shard, and the router holds N cursors per query, so connection-pool pressure and cursor bookkeeping grow with the same factor. ## Step 4: fix the routing, not the capacity In rough order of preference: - **Put the shard key in the filter.** Very often the value is available at the call site but was never passed down — a tenant id, an account id, a user id sitting in the request context but not in the repository method's arguments. Threading it through turns a 20-shard query into a 1-shard query with no data change. - **Reshape the query, not the cluster.** If a screen fetches by a secondary attribute, ask whether the natural entry point is the key-bearing entity instead. - **Serve the shape from a purpose-built collection.** A materialised summary keyed so that the reads become targeted moves the fan-out off the request path onto a background job. - **Accept and contain it.** Some queries genuinely cannot be targeted. Keep them off the latency-critical path, bound their result size, and make sure an index supports them on every shard so each participant's contribution is a short index walk rather than a scan. Resharding the collection on a different key is possible and is the heavier hammer; that decision belongs with shard-key design, and it is worth reaching only when the whole hot workload shares a key that the current one lacks. ## Step 5: keep watching Make fan-out visible in normal operations: track shards-touched for hot shapes after each release, and watch per-shard operation counters relative to application throughput. Broadcast creep is a silent regression — one new filter shipped by one team can turn a well-targeted endpoint into a cluster-wide one without any error, any alert, or any change in the median.