Backpressure and Bounded Queues in Asyncio MQTT Consumers

An asyncio.Queue() created without a maxsize argument is an invitation to an out-of-memory kill at the worst possible moment. This page is the flow-control layer under Async Batch Processing: the specific mechanics of bounding an MQTT consumer’s internal queue so that a broker burst — a reconnect storm redelivering a QoS 1 backlog, a shift-change spike as fifty machines resume simultaneously — degrades the gateway predictably instead of crashing it. The failure this page prevents is subtle because it is invisible until it happens: an unbounded queue works perfectly in every test and every normal shift, then consumes all available memory the first time downstream persistence falls behind ingestion for more than a few minutes, taking the whole process down with it — including the in-flight messages still sitting in the queue.

Producer, bounded queue, and consumer with a full-queue backpressure signal A producer task publishes MQTT messages into a bounded asyncio queue. The consumer task drains the queue in batches. When the queue fills to its configured maxsize, the producer's put operation blocks, which signals backpressure upstream to the MQTT client and pauses new message acceptance until the consumer frees capacity. MQTT producer on_message task Bounded queue asyncio.Queue(maxsize=5000) full: put() blocks Batch consumer drain to writer put() get() batch backpressure: producer paused until room frees consumer rate governs producer rate

The unbounded-queue failure

The default asyncio.Queue() has no size limit, so a producer’s await queue.put(item) never waits — it always succeeds immediately, and the queue grows as long as items arrive faster than the consumer drains them. On a factory floor that gap opens routinely: a TSDB write stalls during a compaction window, a downstream Celery worker pool restarts, or a reconnect delivers a QoS 1 backlog all at once. Under an unbounded queue, none of these conditions produce an error — they produce silent, unbounded memory growth that eventually triggers the Linux OOM killer, which then kills the ingestion process, discarding every message still sitting in the queue. The system fails in exactly the way store-and-forward is designed to prevent, except the failure originates inside the process meant to be protecting the data.

import asyncio

# The anti-pattern: no ceiling, no backpressure signal to the producer.
unbounded_queue: asyncio.Queue = asyncio.Queue()   # maxsize defaults to 0 = unlimited


async def unsafe_on_message(payload: dict) -> None:
    await unbounded_queue.put(payload)   # never blocks; queue grows without limit

There is no code path in the snippet above that would ever notice a consumer falling behind until the process runs out of memory. That is the defect: the absence of a signal, not a bug in any single line.

asyncio.Queue(maxsize) with producer pause/resume

Setting maxsize turns the queue into a real flow-control primitive: once it holds maxsize items, await queue.put(item) blocks the producer coroutine until the consumer removes an item and frees a slot. That block is the backpressure signal — it propagates from the queue back to whatever is calling put, which for an MQTT client means the on_message handler (or the coroutine it schedules) stops accepting new work, which in turn lets the paho-mqtt or gmqtt client’s own internal buffering and TCP flow control take over.

import asyncio
import logging
from dataclasses import dataclass, field

logger = logging.getLogger("bounded_consumer")

QUEUE_MAXSIZE = 5000   # sized to a few seconds of peak throughput, not a guess


@dataclass(slots=True)
class TelemetryMessage:
    asset_id: str
    seq_id: int
    ts: float
    payload: dict


class BoundedMqttConsumer:
    """Bounded asyncio.Queue with explicit pause/resume signaling."""

    def __init__(self, maxsize: int = QUEUE_MAXSIZE):
        self._queue: asyncio.Queue[TelemetryMessage] = asyncio.Queue(maxsize=maxsize)
        self._paused = False

    async def on_message(self, msg: TelemetryMessage) -> None:
        """Called from the MQTT client's message callback (via create_task)."""
        if self._queue.full() and not self._paused:
            self._paused = True
            logger.warning("queue at capacity (%d); producer now blocking on put()",
                           self._queue.maxsize)
        await self._queue.put(msg)   # blocks here until the consumer frees a slot
        if self._paused and not self._queue.full():
            self._paused = False
            logger.info("queue drained below capacity; producer resumed")

    async def get_batch(self, max_items: int, timeout: float) -> list[TelemetryMessage]:
        """Drain up to max_items, or fewer if timeout elapses first."""
        batch: list[TelemetryMessage] = []
        try:
            first = await asyncio.wait_for(self._queue.get(), timeout=timeout)
            batch.append(first)
        except asyncio.TimeoutError:
            return batch
        while len(batch) < max_items:
            try:
                batch.append(self._queue.get_nowait())
            except asyncio.QueueEmpty:
                break
        return batch

    def depth(self) -> int:
        return self._queue.qsize()

The MQTT client library’s own message callback should never block directly — hand the message to on_message via asyncio.create_task or, if the client already runs its network loop inside the event loop (as gmqtt does), await it directly, accepting that a persistently full queue will slow the client’s read loop. That slowdown is the point: it is a controlled, measurable form of degradation instead of an uncontrolled memory leak. Emit queue_depth and a queue_paused_total counter so a dashboard shows exactly how often and for how long the consumer falls behind.

Batch-drain to the writer

Draining one message at a time into the downstream writer wastes the batching efficiency that makes async persistence viable in the first place. get_batch above pulls up to a bound number of items, or fewer if none arrive within a short timeout, so a quiet period still flushes promptly while a busy period amortizes writer overhead across a full batch.

async def drain_loop(consumer: BoundedMqttConsumer, write_batch, *,
                     batch_size: int = 500, flush_timeout: float = 2.0) -> None:
    """Continuously drain the bounded queue into idempotent batch writes."""
    while True:
        batch = await consumer.get_batch(max_items=batch_size, timeout=flush_timeout)
        if not batch:
            continue
        try:
            await write_batch(batch)
        except Exception:
            logger.exception("batch write failed for %d messages; will retry via redelivery",
                             len(batch))
            # Do not re-enqueue here: rely on the broker's QoS 1 redelivery
            # after reconnect rather than doubling memory pressure locally.

Deliberately declining to re-enqueue a failed batch into the same bounded queue matters: retrying locally competes for the same limited capacity that is already under pressure, compounding the problem that caused the failure. Leaning on MQTT’s own QoS 1 at-least-once redelivery — the same mechanism documented in best practices for MQTT QoS levels — means a failed write is recovered by the protocol the system already depends on, not by a second, competing local retry queue.

Shedding vs blocking policy under sustained pressure

Blocking the producer is the correct default response to a transient burst, because it never loses a message — it just delays acceptance until the consumer catches up. But an MQTT client whose on_message blocks indefinitely eventually stalls its own keepalive handling and can trigger a broker-side disconnect, turning a slow-consumer problem into a connection-loss problem. For a sustained backlog — the consumer is stalled for minutes, not seconds — blocking stops being safe and the gateway needs an explicit shedding policy instead, paired with routing the shed messages to the durable edge buffering store-and-forward spool rather than dropping them outright.

import time


class AdaptiveFlowController:
    """Switches from blocking to spill-to-disk once pressure is sustained."""

    def __init__(self, queue: asyncio.Queue, spool_enqueue, *,
                 sustained_pressure_s: float = 15.0):
        self._queue = queue
        self._spool_enqueue = spool_enqueue
        self._sustained_pressure_s = sustained_pressure_s
        self._full_since: float | None = None

    async def offer(self, msg: TelemetryMessage) -> None:
        if self._queue.full():
            self._full_since = self._full_since or time.monotonic()
            if time.monotonic() - self._full_since >= self._sustained_pressure_s:
                # Sustained pressure: stop blocking the MQTT client, spill instead.
                await self._spool_enqueue(msg)
                return
        else:
            self._full_since = None
        await self._queue.put(msg)

This is the same high/low watermark discipline used for ring buffer sizing at the edge, applied to an in-process queue instead of a gateway’s local storage: block first because it is lossless and simple, but have a second line of defense that engages automatically once blocking has clearly stopped being a short-term condition.

Tune sustained_pressure_s against the same worst-case-outage measurement used to size the edge spool, not against a default copied from an unrelated system. A value that is too short spills to disk on every ordinary burst, wearing flash for no durability benefit since blocking alone would have absorbed it; a value that is too long risks the broker-disconnect failure mode above before the escape hatch ever engages. Fifteen seconds is a reasonable starting point for a gateway on a stable LAN; a cellular-backed gateway with higher baseline jitter should measure its own round-trip variance before committing to a number.

Gotchas & anti-patterns

  • Creating asyncio.Queue() with the default maxsize. It silently disables backpressure entirely; always pass an explicit, measured maxsize.
  • Re-enqueueing failed writes into the same bounded queue. This doubles memory pressure exactly when the system is already under pressure. Rely on broker redelivery or the store-and-forward spool instead.
  • Blocking on_message forever with no sustained-pressure escape hatch. An MQTT client that never returns from its callback stalls keepalive processing and can get disconnected by the broker, turning backpressure into connection loss.
  • Sizing maxsize from a guess rather than a measurement. Too small and the queue saturates during ordinary bursts, thrashing between paused and resumed; too large and it delays the OOM problem instead of preventing it. Size from measured peak throughput times the acceptable pause duration.
  • Draining one item at a time. Item-at-a-time draining forfeits the batching efficiency the whole architecture exists to capture; always drain in bounded batches with a flush timeout.

Quick reference

Policy When to use Trade-off
Blocking put() (bounded queue) Transient bursts, seconds-scale Lossless, but can stall MQTT keepalive if sustained
Batch drain with flush timeout Always, alongside either policy above Amortizes writer cost; adds up to flush_timeout latency
Shed to disk spool Sustained pressure beyond tens of seconds Preserves data via store-and-forward; adds disk I/O
Drop (no spool) Never, for OEE-relevant telemetry Silent, unrecoverable loss — avoid in production

Related in this section