# UNDF: (pending)
# elasticsearch-MOADX-0001: EnrichCache.computeIfAbsent — cache stampede on enrich index search

## MOAD-0006 Candidate — The Thundering Herd

| Field | Value |
|-------|-------|
| ID | elasticsearch-MOADX-0001 |
| Severity | HIGH |
| Ecosystem | Elasticsearch (x-pack enrich) |
| File | `x-pack/plugin/enrich/src/main/java/org/elasticsearch/xpack/enrich/EnrichCache.java` lines 94–121 |
| Cache type | Enrich policy result cache (lookup index search results) |
| Pattern | get → null → searchResponseFetcher (Elasticsearch search query) → put (explicitly non-locking per comment) |
| Trigger | N concurrent ingest pipeline requests all miss the same enrich cache key → N×Elasticsearch search queries fired simultaneously |
| Hot path | Every document ingested through an enrich processor that misses cache |

## Defect

`EnrichCache.computeIfAbsent` (method name misleadingly implies atomicity) performs a raw get-then-execute-then-put without any synchronization. The code's own comment acknowledges this:

```java
public void computeIfAbsent(
    ProjectId projectId,
    String enrichIndex,
    Object lookupValue,
    int maxMatches,
    Consumer<ActionListener<SearchResponse>> searchResponseFetcher,
    ActionListener<List<Map<?, ?>>> listener
) {
    // intentionally non-locking for simplicity...it's OK if we re-put the same key/value in the cache during a race condition.
    long cacheStart = relativeNanoTimeProvider.getAsLong();
    var cacheKey = new CacheKey(projectId, enrichIndex, lookupValue, maxMatches);
    List<Map<?, ?>> response = get(cacheKey);           // (1) all N threads: miss
    long cacheRequestTime = relativeNanoTimeProvider.getAsLong() - cacheStart;
    if (response != null) {
        hitsTimeInNanos.add(cacheRequestTime);
        listener.onResponse(response);
    } else {
        final long retrieveStart = relativeNanoTimeProvider.getAsLong();
        searchResponseFetcher.accept(ActionListener.wrap(resp -> {  // (2) N×search queries
            CacheValue cacheValue = toCacheValue(resp);
            put(cacheKey, cacheValue);                  // (3) N redundant puts
            // ...
        }, listener::onFailure));
    }
}
```

The comment "it's OK if we re-put the same key/value" is wrong for high-cardinality enrich lookups. It is not OK when:
1. **The enrich index is large**: each search hits the Lucene index for that enrich data, consuming heap and I/O
2. **The cache key has high cardinality** (e.g., IP address lookup in a geo-enrich policy): TTL expiry under ingestion load causes N threads to all fire searches for the same IP simultaneously
3. **The search requires network hops**: in a multi-node cluster the enrich coordinator proxies searches to the shard owner (see `EnrichCoordinatorProxyAction`) — N concurrent searches all go to the same shard

### Ingest pipeline throughput amplification

Enrich processors are designed to be on the hot path for bulk indexing. At 10,000 documents/second with a TTL-expired cache entry, the cluster can receive 10,000 simultaneous search queries for the same lookup value before the first one completes and populates the cache.

## Impact

- Enrich index shard becomes saturated with redundant identical searches
- Ingest throughput drops dramatically during cache cold start or TTL expiry
- In worst case (hot key + small TTL), the stampede is continuous: searches pile up before cache is populated, the pipeline backs up, and ES node heap fills with queued search contexts
- Kibana's `_enrich/stats` will show abnormally high `misses_time_in_nanos` with ratio misses/hits > 1

## Fix

Use a `ConcurrentHashMap` of in-flight `CompletableFuture`s to coalesce concurrent misses for the same key — the same pattern as Pulsar's `BookkeeperSchemaStorage.readSchemaOperations`:

```java
private final ConcurrentHashMap<CacheKey, CompletableFuture<List<Map<?, ?>>>> inflight =
    new ConcurrentHashMap<>();

public void computeIfAbsent(
    ProjectId projectId,
    String enrichIndex,
    Object lookupValue,
    int maxMatches,
    Consumer<ActionListener<SearchResponse>> searchResponseFetcher,
    ActionListener<List<Map<?, ?>>> listener
) {
    var cacheKey = new CacheKey(projectId, enrichIndex, lookupValue, maxMatches);
    List<Map<?, ?>> cached = get(cacheKey);
    if (cached != null) {
        listener.onResponse(cached);
        return;
    }
    // Coalesce: only one search fires per key; all other waiters share the future
    CompletableFuture<List<Map<?, ?>>> future = inflight.computeIfAbsent(cacheKey, k -> {
        CompletableFuture<List<Map<?, ?>>> f = new CompletableFuture<>();
        searchResponseFetcher.accept(ActionListener.wrap(resp -> {
            CacheValue cacheValue = toCacheValue(resp);
            put(cacheKey, cacheValue);
            inflight.remove(cacheKey);
            f.complete(deepCopy(cacheValue.hits, false));
        }, ex -> {
            inflight.remove(cacheKey);
            f.completeExceptionally(ex);
        }));
        return f;
    });
    future.whenComplete((result, ex) -> {
        if (ex != null) {
            listener.onFailure(new RuntimeException(ex));
        } else {
            listener.onResponse(result);
        }
    });
}
```

Alternatively, replace the custom `Cache` with Caffeine `AsyncLoadingCache` which provides this deduplication automatically.
