Backpressure and Admission Control

Every distributed data system eventually meets the same wall: more work arrives than it can do. The naive response — accept everything and let queues grow — is how clusters die. Queues consume heap, heap pressure trips circuit breakers, garbage collection stalls, nodes drop out of the cluster, shards relocate onto already-overloaded nodes, and a localized hotspot becomes a cascading failure that takes down the whole cluster. The single overloaded shard kills everything.

Backpressure and admission control is the disciplined alternative: when a node or a shard is overloaded, reject some work early and cheaply — with a clear, retryable signal — so the system sheds load instead of collapsing. OpenSearch does this on both sides of the workload: shard indexing pressure (smart, per-shard rejection of writes) and search backpressure (resource tracking and cancellation of expensive queries). The art is rejecting the right work: the request actually responsible for the pressure, not an innocent bystander.

After this chapter you should be able to: explain why naive queueing causes cascading failure; describe how shard indexing pressure tracks outstanding bytes and rejects per-shard; explain secondary-parameter rejection; describe search backpressure's task-resource-tracking and cancellation model; and cite the real RFC trail.

Prerequisites. This is the companion to circuit breakers and memory (the last line of defense) — backpressure is the earlier, smarter line. It also touches threadpools and concurrency.

Why naive queueing kills clusters

flowchart TD
    Load["write load > capacity on shard X"] --> Q["accept everything → queues grow"]
    Q --> Heap["heap fills with in-flight requests"]
    Heap --> CB["circuit breaker trips / GC stalls"]
    CB --> Drop["node drops out of cluster"]
    Drop --> Reroute["shards relocate onto other overloaded nodes"]
    Reroute --> Spread["overload spreads → cascading failure"]

The failure is positive feedback: overload causes a node to leave, which causes its shards to pile onto survivors, which overloads them, which causes them to leave. Backpressure breaks the loop at the first step by refusing to accept work it cannot do, returning a 429 TOO_MANY_REQUESTS that a well-behaved client retries with backoff. A rejected-and-retried request is recoverable; a dead cluster is not.

Note: Circuit breakers (memory accounting that trips when a single operation would exceed a heap limit) are the last line of defense — they prevent OOM but are blunt. Backpressure is earlier and smarter: it watches sustained pressure and rejects the specific shard/request causing it, before the breaker ever needs to fire. Read circuit-breakers-memory.md for the relationship.

Shard indexing pressure: smart, per-shard rejection

The key OpenSearch insight (from the shard-level backpressure RFC) is that node-level rejection is too coarse. A node hosts many shards; one hot shard being slammed should not cause the node to reject writes destined for other, healthy shards on the same node. So OpenSearch tracks indexing pressure per shard and rejects only the writes hitting the shard that is actually in trouble.

The mechanism tracks outstanding bytes of in-flight indexing work, at multiple stages of the write path:

When a shard's outstanding bytes exceed its limit, new writes to that shard are rejected — leaving other shards on the node unaffected.

flowchart TD
    W["write op for shard X"] --> Acct["account bytes against shard X (coordinating/primary/replica)"]
    Acct --> Check{shard X over limit?}
    Check -->|no| Accept["proceed; release bytes on completion"]
    Check -->|yes| Sec{secondary parameters also bad?}
    Sec -->|yes| Reject["reject: 429, this shard only"]
    Sec -->|no| Accept

Secondary-parameter rejection

A pure byte threshold over-rejects: a shard can briefly hold a lot of bytes during a healthy burst without being in trouble. So shard indexing pressure does not reject on the byte limit alone — it consults secondary parameters (signals that the shard is genuinely not keeping up, such as the throughput degrading or the request queue/latency for that shard rising) before rejecting. The intent is to reject only shards that are both over their byte budget and showing signs of real distress — minimizing false rejections of healthy bursts.

cd ~/src/OpenSearch
# The shard indexing pressure tracker + the rejection logic.
grep -rln "ShardIndexingPressure\|IndexingPressure\|ShardIndexingPressureStore" \
  server/src/main/java/org/opensearch/index/ | head
grep -rn "coordinatingBytes\|primaryBytes\|replicaBytes\|secondary\|rejection\|markCoordinating" \
  server/src/main/java/org/opensearch/index/stats/ \
  server/src/main/java/org/opensearch/index/ 2>/dev/null | head -20
# The settings.
grep -rn "shard_indexing_pressure\|indexing_pressure" \
  server/src/main/java/org/opensearch/ | head

Search backpressure: track resources, cancel the offender

Reads need protection too, but the model is different. You cannot "reject bytes" — an expensive search has already been accepted before you know it is expensive (a deep aggregation, a giant terms, a runaway script). So search backpressure tracks the resource consumption of in-flight search tasks and cancels the ones consuming disproportionately when the node is under strain.

flowchart TD
    Node["node under resource strain (heap/CPU)"] --> Track["task resource tracking: per-task CPU + heap"]
    Track --> Rank["rank in-flight search tasks by consumption"]
    Rank --> Pick["pick the worst offenders over thresholds"]
    Pick --> Cancel["cancel those tasks (TaskCancellation)"]
    Cancel --> Relief["node sheds load; healthy queries continue"]

The building blocks: the task framework assigns every search a cancellable task; task resource tracking records its CPU time and heap allocations; a backpressure service periodically evaluates node strain and, when over threshold, cancels the tasks consuming the most — so one runaway query dies instead of the node.

# Search backpressure service + task resource tracking.
ls server/src/main/java/org/opensearch/search/backpressure/ 2>/dev/null
grep -rln "SearchBackpressureService\|SearchBackpressureSettings\|TaskResourceTracking\|CancellableTask" \
  server/src/main/java/org/opensearch/search/backpressure/ \
  server/src/main/java/org/opensearch/tasks/ 2>/dev/null | head
grep -rn "search_backpressure\|cancellation" \
  server/src/main/java/org/opensearch/search/backpressure/ 2>/dev/null | head

The search-backpressure capstone is about adding a new signal to this evaluation — which forces you to understand the whole tracking-and-cancellation loop.

The RFC trail

Read #478 for the why per-shard, then #1336 for the how — it is a clean example of an RFC's design becoming concrete tracking code.

Settings (grep for the real keys)

# Confirm exact keys/defaults in your checkout.
grep -rn "shard_indexing_pressure\|search_backpressure" \
  server/src/main/java/org/opensearch/ | grep -i "Setting" | head -20

# Observe rejections / backpressure stats live.
curl -s 'localhost:9200/_nodes/stats/indexing_pressure?pretty' 2>/dev/null | head -40
curl -s 'localhost:9200/_nodes/stats/search_backpressure?pretty' 2>/dev/null | head -40

Try it / read it

# 1. Read the trackers.
grep -rln "ShardIndexingPressure" server/src/main/java/org/opensearch/index/ | head
ls server/src/main/java/org/opensearch/search/backpressure/

# 2. Enable enforcement, then hammer one shard and watch for 429s.
curl -s -XPUT 'localhost:9200/_cluster/settings' -H 'Content-Type: application/json' -d '{
  "persistent": {
    "shard_indexing_pressure.enabled": true,
    "shard_indexing_pressure.enforced": true
  }
}'
# generate concurrent bulk load to one index, then check stats:
curl -s 'localhost:9200/_nodes/stats/indexing_pressure?pretty' | grep -i "rejection\|current\|limit" | head

# 3. Enable search backpressure and run a deliberately expensive query.
curl -s -XPUT 'localhost:9200/_cluster/settings' -H 'Content-Type: application/json' \
  -d '{"persistent":{"search_backpressure.mode":"enforced"}}'
curl -s 'localhost:9200/_nodes/stats/search_backpressure?pretty' | grep -i "cancel\|limit" | head

# 4. Tests.
./gradlew :server:test --tests "*ShardIndexingPressure*" 2>&1 | tail -15
./gradlew :server:test --tests "*SearchBackpressure*" 2>&1 | tail -15

Common symptoms (and what they mean)

Validation: prove you understand this

1. Draw the cascading-failure loop and mark the step where backpressure breaks it.
2. Explain why per-shard rejection is better than per-node, with a concrete two-shard example.
3. Name the three byte quantities shard indexing pressure tracks and why secondary-parameter rejection exists (what failure does it prevent?).
4. Contrast indexing backpressure (reject bytes early) with search backpressure (track + cancel) and explain why reads need a different model.
5. Describe the search backpressure loop: task → resource tracking → evaluation → cancellation, naming the framework pieces.
6. Explain the relationship between backpressure and circuit breakers: which fires first and why backpressure is "smarter".

Next: Star-Tree Indexes and Aggregation Acceleration, or the search-backpressure capstone.