Backpressure with Bounded asyncio Queues #

An SNMP trap storm from a failed optical amplifier does not politely wait for the collector to catch up. A single protected transport ring can emit 40,000 traps in the first two seconds of a fault, and the moment a collector reads them faster than the correlation engine can drain them, those events have to live somewhere. With an unbounded asyncio.Queue, that somewhere is the process heap: the queue grows without limit, resident memory climbs past the container’s cgroup ceiling, and the orchestrator OOM-kills the pod mid-storm — exactly when the NOC most needs telemetry. The restart loses every buffered event, the collector reconnects into the same storm, and the cycle repeats. The result is not a slow pipeline; it is a pipeline that vanishes for the duration of the outage, adding minutes of blind time straight onto mean time to acknowledge (MTTA) and mean time to resolution (MTTR).

Backpressure is the discipline that stops this. Instead of letting producers push at line rate into infinite memory, a bounded queue makes the slowest stage in the pipeline set the pace for every stage upstream of it. When the consumer falls behind, the queue fills, and the fill signal propagates backward until the producer either slows down or makes an explicit, observable decision to shed load. The choice between those two responses — block the producer or drop the event — is the single most important policy decision in a storm-resilient collector.

Schema alignment and taxonomy anchor #

This page is the flow-control primitive underneath Async Batch Processing, the stage of the Ingestion & Parsing Workflows data plane that aggregates normalized fault records into correlation envelopes. That parent stage already relies on bounded asyncio.Queue instances per severity tier to convert ingest pressure into an explicit drop decision; this reference zooms into how that boundedness is chosen, measured, and enforced end to end so the accumulator upstream of it never becomes the thing that runs out of memory.

Every event crossing the queue is a normalized record conforming to the canonical contract from Event Schema Design, carrying its node_id, severity, and vendor_alarm_code. Backpressure never inspects those fields to decide flow — that would couple transport to semantics — but it does read severity for one purpose only: when it must shed, it sheds the lowest tier first, so a bounded queue under pressure degrades the same way the rest of the pipeline does. The protocol-specific tuning for the highest-volume source class is worked through in Implementing Asyncio for High-Volume SNMP, which this reference composes into a general flow-control pattern.

A bounded queue with high-water and low-water marks #

The core pattern is a single bounded asyncio.Queue with a maxsize sized to a fixed memory budget, plus two thresholds that give the system hysteresis. A high-water mark (say 80% of maxsize) is where the producer starts throttling; a low-water mark (say 40%) is where it resumes full speed. Using one threshold for both directions causes flapping — the producer toggles on and off every few events at the boundary — so the gap between the two marks is deliberate.

import asyncio
import logging
import time
from dataclasses import dataclass, field
from enum import IntEnum
from typing import Any, Optional

from pydantic import BaseModel, ConfigDict, Field, field_validator

logger = logging.getLogger("backpressure")


class Severity(IntEnum):
    CRITICAL = 1
    MAJOR = 2
    MINOR = 3
    WARNING = 4


class FaultEvent(BaseModel):
    # Strict mode keeps a malformed trap from silently coercing into a valid event.
    model_config = ConfigDict(strict=True, extra="forbid")

    node_id: str
    severity: Severity
    vendor_alarm_code: str
    event_time: float = Field(gt=0)
    payload: dict[str, Any] = Field(default_factory=dict)

    @field_validator("event_time")
    @classmethod
    def reject_future_skew(cls, v: float) -> float:
        # A timestamp far in the future is a clock-drift artifact, not a real event.
        if v > time.time() + 5.0:
            raise ValueError("event_time exceeds allowable clock skew")
        return v


class SheddingPolicy(IntEnum):
    BLOCK = 0   # slow the producer: await put() until the consumer drains
    SHED = 1    # drop the lowest tier: never block the read loop


@dataclass
class BoundedIngest:
    maxsize: int = 20_000
    high_water: float = 0.80      # fraction of maxsize that trips throttling
    low_water: float = 0.40       # fraction that releases it
    policy: SheddingPolicy = SheddingPolicy.SHED
    shed_below: Severity = Severity.MINOR  # only Minor/Warning are ever shed
    queue: asyncio.Queue = field(init=False)
    _throttled: bool = field(default=False, init=False)
    metrics: dict[str, int] = field(
        default_factory=lambda: {"enqueued": 0, "shed": 0, "blocked": 0}
    )

    def __post_init__(self) -> None:
        self.queue = asyncio.Queue(maxsize=self.maxsize)

    def _depth_fraction(self) -> float:
        return self.queue.qsize() / self.maxsize

    def _update_throttle(self) -> None:
        # Hysteresis: trip at the high mark, release only at the low mark.
        frac = self._depth_fraction()
        if not self._throttled and frac >= self.high_water:
            self._throttled = True
            logger.warning("backpressure engaged at depth=%.0f%%", frac * 100)
        elif self._throttled and frac <= self.low_water:
            self._throttled = False
            logger.info("backpressure released at depth=%.0f%%", frac * 100)

    async def offer(self, event: FaultEvent) -> bool:
        """Enqueue one event, applying the configured backpressure policy."""
        self._update_throttle()

        if self._throttled and self.policy is SheddingPolicy.SHED:
            # Shed only low-severity noise; Critical/Major always get a slot.
            if event.severity >= self.shed_below:
                self.metrics["shed"] += 1
                return False

        if self.policy is SheddingPolicy.BLOCK:
            # await put() is the backpressure: it suspends the producer coroutine
            # until the consumer has made room, so memory can never run away.
            if self.queue.full():
                self.metrics["blocked"] += 1
            await self.queue.put(event)
        else:
            try:
                self.queue.put_nowait(event)
            except asyncio.QueueFull:
                # Even in SHED mode a Critical event that finds a full queue
                # must not be lost: fall back to an awaited put for it alone.
                if event.severity <= Severity.MAJOR:
                    await self.queue.put(event)
                else:
                    self.metrics["shed"] += 1
                    return False

        self.metrics["enqueued"] += 1
        return True

The pivotal line is await self.queue.put(event). When the queue is full, that coroutine does not return until the consumer has called get() and freed a slot. Suspending there is the backpressure: the producer’s read loop stops pulling from the socket, the OS receive buffer fills, and the pressure propagates all the way back to the sender — no memory is allocated to hold events the pipeline cannot yet process. The shed_below guard guarantees that even when the policy is to drop, only Minor and Warning events are ever dropped; Critical and Major always take the awaited-put path.

Async processing hook #

Backpressure is only end to end if the producer actually respects the throttle instead of busy-spinning. The producer loop below reads its throttle state before each read burst, and the consumer signals progress purely by draining the queue — no cross-coroutine locks are needed because a single-threaded event loop serializes the qsize() reads.

async def producer(ingest: BoundedIngest, source: asyncio.Queue) -> None:
    """Read normalized traps and offer them under backpressure."""
    while True:
        raw = await source.get()          # upstream socket/decoder handoff
        try:
            event = FaultEvent.model_validate(raw)
        except Exception:
            source.task_done()
            continue                       # malformed → dead-letter upstream
        # If throttled in BLOCK mode, offer() awaits and naturally paces us.
        await ingest.offer(event)
        source.task_done()


async def consumer(ingest: BoundedIngest, accumulator) -> None:
    """Drain the bounded queue into the batch accumulator, never blocking on it."""
    while True:
        event = await ingest.queue.get()
        try:
            await accumulator.add(event)   # awaited handoff downstream
        finally:
            ingest.queue.task_done()


async def run_pipeline(source: asyncio.Queue, accumulator) -> None:
    ingest = BoundedIngest(maxsize=20_000, policy=SheddingPolicy.SHED)
    async with asyncio.TaskGroup() as tg:
        tg.create_task(producer(ingest, source))
        # Several consumers drain in parallel; more consumers = a higher
        # low-water release rate, which is the real cure for chronic backpressure.
        for _ in range(4):
            tg.create_task(consumer(ingest, accumulator))

Running several consumers against one bounded queue is the direct lever on how often backpressure engages: throughput out of the queue is the consumer count multiplied by per-item drain time, and raising it lowers the steady-state depth. The producer count stays at one per source socket, because the point is to let the consumers set the pace.

Mitigation and hardening #

When the pipeline is saturated the system must degrade predictably rather than stall or crash. These are the concrete failure paths a production collector should implement:

1. Shed-versus-block by severity, never globally. Blocking the producer is correct for a batch pipeline where the source can wait (a Kafka partition holds offsets), but wrong for a live UDP trap socket where a blocked reader means kernel buffer overflow and silent kernel-level drops. Choose BLOCK when the upstream is replayable and SHED when it is lossy at the wire. In both cases route shed events, with their node_id and vendor_alarm_code, to a dead-letter queue so the drop is countable, not invisible.
2. Never shed Critical. The shed_below floor is the inviolable guarantee: a SERVICE_OUTAGE trap takes the awaited-put path even under a full queue, mirroring the priority-bypass posture used in Rate Limiting Strategies. Losing a P1 during a storm is the one outcome backpressure exists to prevent.
3. Cap the awaited-put wait. An awaited put() that never returns because every consumer has died is a deadlock. Wrap it in asyncio.wait_for() with a timeout equal to a few window periods; on timeout, spill the event to a disk-backed overflow buffer and page, because a stall this long means the consumers, not the queue, are the fault.
4. Watch depth, not just drops. A drop counter climbing is a symptom; queue depth crossing the high-water mark is the leading indicator. Alert on sustained time-above-high-water, which predicts shedding before a single event is lost.

Operational hardening notes #

Tuning backpressure is really tuning one number against a memory budget: maxsize multiplied by the average serialized event size must stay comfortably under the container’s memory limit, leaving headroom for the batch accumulator and the interpreter itself. A 20,000-slot queue of roughly 1 KB events is about 20 MB of worst-case resident data — a deliberate, bounded cost you can reason about, unlike an unbounded queue’s unbounded one. Keep the high-water and low-water marks at least 30 percentage points apart so the throttle does not flap at the boundary during a steady storm. Read queue.qsize() for observability, but never branch correlation logic on it, because it is a point-in-time sample that a concurrent get() can invalidate a microsecond later; the authoritative flow control is the awaited put() itself, which is race-free by construction on a single event loop. Finally, size the consumer pool from the drain-time math rather than by guessing: if a single consumer drains 5,000 events per second and the storm peaks at 20,000 per second, four consumers hold the line and a fifth gives margin. Held to these constraints, a bounded queue turns an unbounded, unobservable OOM risk into a bounded, alertable, sub-2-second ticket-creation budget that survives the worst trap storm the network can produce.

Frequently Asked Questions #

Why does a bounded queue prevent an out-of-memory kill when an unbounded one does not? #

An unbounded asyncio.Queue accepts every put_nowait immediately, so during a storm it grows until resident memory exceeds the container limit and the orchestrator kills the process. A bounded queue has a fixed maxsize, so an awaited put() suspends the producer once the queue is full instead of allocating more memory. That suspension is the backpressure: it caps worst-case memory at maxsize times the event size, a number you choose deliberately against your memory budget.

When should I block the producer instead of shedding events? #

Block when the upstream source is replayable and can wait without losing data, such as a Kafka partition that simply holds its offset while the producer is suspended. Shed when the source is lossy at the wire, such as a live UDP trap socket where a blocked reader causes kernel buffer overflow and uncounted drops. Blocking preserves every event but pushes pressure upstream; shedding drops the lowest severity tier but keeps the read loop moving.

Why use two water marks instead of one threshold? #

A single threshold makes the producer toggle throttling on and off every few events right at the boundary, which wastes cycles and produces noisy metrics. A high-water mark that engages the throttle and a lower low-water mark that releases it give the system hysteresis, so the producer stays throttled through the busy period and only resumes full speed once the consumer has drained a real buffer of headroom. Keep the two marks at least thirty percentage points apart.

How do I guarantee Critical alarms are never shed under backpressure? #

Set a severity floor below which shedding is allowed and keep Critical and Major above it. When the policy would otherwise drop an event, the code checks severity first: Minor and Warning events are dropped and counted, while Critical and Major fall through to an awaited put that suspends the producer until a slot frees rather than losing the event. Because the awaited put cannot lose data, a P1 outage trap always reaches the correlation engine even when the queue is full.

Related #

• Up to: Async Batch Processing — the accumulation stage this bounded queue feeds and protects
• Implementing Asyncio for High-Volume SNMP — protocol-specific tuning for the highest-volume producer class
• Rate Limiting Strategies — admission control that sheds noise before it reaches this queue
• Error Categorization Pipelines — where dead-lettered shed events are triaged
• Event Schema Design — the canonical contract every enqueued event satisfies