Distributed Search Engine: Latency Waterfall
Break down end-to-end latency by hop and percentile. Understand where the p99 tail comes from — DNS, network, cache lookup, database query, serialization. Real requests have wildly different paths depending on cache-hit location.
Latency Waterfall
Break down end-to-end latency by hop (network, application, database, cache) and percentile (p50/p90/p99/p99.9). Real requests have wildly different paths depending on cache-hit location — pick a scenario to see the full waterfall.
Amara Google 2009: every 100ms of latency = 1% revenue lost. Understanding WHERE the tail comes from is the difference between random optimization and targeted engineering.
Fast query — all shards respond quickly (~85% of requests)
Coordinator fans out query to 32 shards. All shards respond within their normal latency budget. Coord merges top-K globally, returns.
Client → ALB → coord routing. Same-region hop, TLS session resumed. Coordinator node is separate role (not data node).
Parse query DSL, choose shards to query (all 32 for keyword search, fewer for filtered), pick replica per shard via adaptive replica selection (ARS picks fastest replica based on recent latency).
Optimize: Adaptive replica selection eliminates known-slow replicas. Configure `cluster.routing.use_adaptive_replica_selection: true`.
Fire-and-forget queries to all 32 shard replicas in parallel. Latency here is TCP send, not response. Response latency is tracked per shard below.
Each shard runs Lucene BM25 on local inverted index. Doc counts + top-K per shard computed. Tail dominated by JVM GC pauses, cold cache, disk seeks.
Optimize: Warm caches via `pre-fetch` on cold indices. Configure JVM heap 30GB (below compressed-oops boundary). Use SSD (gp3 min).
Coord blocks until ALL shards respond (or hedged request wins). In this scenario, all shards fast — coord returns quickly. This hop reflects the 'wait for tail' cost.
Merge 32 sorted top-K lists into global top-K using heap-based merge. Complexity O(N log K) where N = 32 * K.
Elasticsearch two-phase query: (1) query phase returns doc IDs + scores per shard, (2) fetch phase retrieves documents. If _source stored, single hop.
Serialize JSON response with hits array.
Fast query hits the sub-50ms sweet spot. **Notice the shard latency (p99 40ms) DOMINATES total**. Even the coord + network overhead is a small fraction. The takeaway: shard-level Lucene execution is the actual work, everything else is orchestration.
Bottleneck summary
Distributed search latency is DOMINATED BY THE SLOWEST SHARD (scatter-gather = coord waits for max, not average). **Hedged requests transform the p99.9 tail** by racing primary against replica when primary is slow. Cost: ~5% extra query load. Benefit: p99.9 drops 10-30x. Fast queries (85% of traffic) hit sub-50ms. Slow-shard queries (10%) recover via hedging to ~100ms. Cold-shard queries (5%) also recover via hedging. **Without hedged requests, p99.9 would be dominated by rare GC pauses + cold caches — user-visible latency variance.**
Optimization tips (this architecture)
- **Enable adaptive replica selection**: `cluster.routing.use_adaptive_replica_selection: true`. ARS picks fastest replica based on recent stats — eliminates known-slow replicas.
- **Tune hedged request threshold**: Set to p95 of shard latency. Elasticsearch does this automatically; verify via `_search` profile.
- **JVM heap 30GB**: Below compressed-oops boundary. Larger heap = longer GC pauses. Use G1GC or ZGC for shorter pauses.
- **Warm cold shards**: Run synthetic queries after rolling deploys. Use `_cache/clear` sparingly (only when memory pressure). Pre-fetch with `_forcemerge`.
- **Separate coord + data nodes**: 3 dedicated coord nodes handle query aggregation; 6+ data nodes handle Lucene. Prevents coord GC affecting data nodes.
- **Query DSL efficiency**: `filter` for non-scoring, `query` for scoring. `constant_score` for boolean filters. Avoid `wildcard` (except leading-anchored), `regex`.
- **Two-phase query**: Return only doc_ids in query phase, fetch _source in second phase. Reduces payload if _source is large.
- **Refresh interval tuning**: Default 1s makes new docs searchable but hurts indexing throughput. Increase to 30s for write-heavy indices.
Where to go next
Now that you can see where latency comes from, trace how the architecture EVOLVES to handle 10x more traffic. Or dive into the masterclass for the full ADR + business exercise + incident narrative.