Backpressure-Aware Async Ingestion Queues in Python asyncio

This page solves one specific production failure: a burst of telemetry arrives faster than the batch writer can commit it, the in-memory buffer grows without limit, and the ingestion worker is killed by the OOM killer mid-shift — taking the unwritten tail of the stream with it.

The fix is not a bigger machine; it is a bounded asyncio.Queue sitting between the network reader and the batch writer so that when the writer falls behind, the reader is made to wait rather than allowed to allocate. That waiting is backpressure: an unbounded queue converts a slow consumer into unbounded memory, while a bounded queue converts a slow consumer into a slow producer, which is survivable. This queue is the flow-control stage of the Async Batch Processing layer, and it sits downstream of the coordinate checks in Validating GPS Coordinates in Python — validated fixes enter the queue, chunked batches leave it for the solver.

Environment & Data Prerequisites

The queue and workers use only the Python 3.10+ standard library — asyncio, logging, json, dataclasses, and datetime. The single third-party dependency is the async test harness, needed only for the verification step:

python -m pip install pytest==8.2.0 pytest-asyncio==0.23.8

Each item that flows through the queue is one validated telemetry record. The worker reasons over exactly these fields:

Two design rules matter before any code. First, the queue is bounded — maxsize is a hard cap, not a hint, and the producer must reach it through await queue.put(...) so the event loop can suspend it. Second, nothing is ever dropped on the ingestion path: dropped is an invariant that must read zero at end of run, which is why the producer blocks rather than discards.

Step-by-Step Implementation

Step 1 — Structured commit-audit logging

Every batch commit and every shutdown must be machine-readable and replayable, so decisions are emitted as single-line JSON through a custom formatter. This mirrors the framework’s JSONFormatter pattern, but its fields are specific to flow control — queue depth and batch size, not coordinates.

import asyncio
import json
import logging
from dataclasses import dataclass, asdict, field
from datetime import datetime, timezone
from typing import Optional


class JSONFormatter(logging.Formatter):
    """Emit each log record as a single-line JSON object for ingestion audit."""

    def format(self, record: logging.LogRecord) -> str:
        entry = {
            "logged_at": datetime.now(timezone.utc).isoformat(),
            "level": record.levelname,
            "event": record.getMessage(),
            "payload": getattr(record, "payload", None),
        }
        return json.dumps(entry)


audit_logger = logging.getLogger("waste_ops.ingest_queue")
if not audit_logger.handlers:
    _handler = logging.StreamHandler()
    _handler.setFormatter(JSONFormatter())
    audit_logger.addHandler(_handler)
audit_logger.setLevel(logging.INFO)

Step 2 — The record and the bounded queue

Each item is a frozen dataclass, and the queue is constructed with an explicit maxsize. A maxsize of zero means unbounded in asyncio, which is precisely the failure mode this page exists to prevent — so the constructor forbids it.

@dataclass(frozen=True)
class TelemetryRecord:
    device_id: str
    seq: int
    ts: str
    fill_pct: float


def make_bounded_queue(maxsize: int) -> "asyncio.Queue[Optional[TelemetryRecord]]":
    if maxsize <= 0:
        raise ValueError("maxsize must be > 0; a zero maxsize is unbounded and defeats backpressure")
    return asyncio.Queue(maxsize=maxsize)

Step 3 — A producer that awaits on put (backpressure)

The producer reads the upstream stream and hands each record to the queue with await queue.put(...). When the queue is full the await suspends the producer until the consumer frees a slot — that suspension is the backpressure. When the stream ends, the producer enqueues one None sentinel per consumer so the consumers know to drain and stop.

async def produce(
    queue: "asyncio.Queue[Optional[TelemetryRecord]]",
    stream: list[TelemetryRecord],
    consumers: int,
) -> None:
    for record in stream:
        await queue.put(record)          # suspends here when the queue is full — backpressure
    for _ in range(consumers):
        await queue.put(None)            # one sentinel per consumer to signal end-of-stream

Step 4 — A consumer that drains in chunked batches

The consumer pulls one item with await queue.get() (which blocks it when the queue is empty), then greedily drains whatever else is already queued with get_nowait() up to a batch size. It commits the batch, calls task_done() once per item so queue.join() can settle, and emits the audit record. On the sentinel it flushes whatever partial batch it holds — this is the graceful drain that protects the tail.

async def consume(
    queue: "asyncio.Queue[Optional[TelemetryRecord]]",
    batch_size: int,
    committed: list[TelemetryRecord],
) -> None:
    batch: list[TelemetryRecord] = []

    async def commit(reason: str) -> None:
        if not batch:
            return
        committed.extend(batch)          # stands in for the durable batch writer
        audit_logger.info(
            "batch_committed",
            extra={"payload": {
                "reason": reason,
                "batch_size": len(batch),
                "queue_depth": queue.qsize(),
                "dropped": 0,
            }},
        )
        batch.clear()

    while True:
        item = await queue.get()         # suspends here when the queue is empty
        if item is None:                 # end-of-stream sentinel
            await commit("shutdown_drain")
            queue.task_done()            # balance the get() for the sentinel itself
            return
        pending = [item]
        stop = False
        # greedily absorb anything already waiting, without ever blocking
        while len(pending) < batch_size:
            try:
                nxt = queue.get_nowait()
            except asyncio.QueueEmpty:
                break
            if nxt is None:              # sentinel surfaced mid-drain: flush then stop
                queue.task_done()        # balance the get_nowait() for the sentinel
                stop = True
                break
            pending.append(nxt)
        batch.extend(pending)
        await commit("shutdown_drain" if stop else "batch_full")
        for _ in pending:                # exactly one task_done per real item consumed
            queue.task_done()
        if stop:
            return

The task_done() accounting balances get() exactly once per item — including the sentinels — so queue.join() would settle cleanly. The runnable orchestration in Step 5 uses a single consumer and awaits the consumer task directly, which is simpler than join() when the sentinel already signals completion.

Step 5 — Wire it up and run a burst

The orchestration builds a small bounded queue, floods it with a burst far larger than maxsize, and lets backpressure throttle the producer to the consumer’s pace. No item is dropped; the tail is flushed on shutdown.

async def run_ingestion(maxsize: int = 8, batch_size: int = 4) -> list[TelemetryRecord]:
    queue = make_bounded_queue(maxsize)
    committed: list[TelemetryRecord] = []

    burst = [
        TelemetryRecord("bin-A17", seq, f"2026-07-16T09:{seq // 60:02d}:{seq % 60:02d}+00:00", 40.0 + seq)
        for seq in range(50)             # 50 records through an 8-slot queue
    ]

    consumer = asyncio.create_task(consume(queue, batch_size, committed))
    await produce(queue, burst, consumers=1)
    await consumer                       # graceful shutdown: consumer returns after draining
    return committed


if __name__ == "__main__":
    result = asyncio.run(run_ingestion())
    print(f"ingested {len(result)} records, dropped 0")

Compliance Output

Each committed batch emits one JSON line to the append-only ingestion audit. A batch_full commit and the final shutdown_drain commit from the run above serialize as:

{"logged_at": "2026-07-16T09:03:11.402118+00:00", "level": "INFO", "event": "batch_committed", "payload": {"reason": "batch_full", "batch_size": 4, "queue_depth": 8, "dropped": 0}}
{"logged_at": "2026-07-16T09:03:11.409552+00:00", "level": "INFO", "event": "batch_committed", "payload": {"reason": "shutdown_drain", "batch_size": 2, "queue_depth": 0, "dropped": 0}}

Each field earns its place in the audit trail:

• reason — batch_full for a size-triggered commit versus shutdown_drain for the final partial flush. A run that ends without exactly one shutdown_drain is proof the tail was lost.
• batch_size — the number of records in this commit. Summed across every line it must equal the record count the producer accepted, which is how “no records lost” is audited after the fact.
• queue_depth — queue.qsize() at commit time. A depth pinned at maxsize across the run is the signature of a healthy backpressure regime: the queue is full because the reader is being throttled, not because memory is growing.
• dropped — held at 0 as an explicit invariant. Under municipal collection-record rules a missed service reading cannot be silently discarded; the backpressure design guarantees the field by construction, and logging it makes that guarantee inspectable.

Because the ledger is append-only and every commit records its own depth and size, the trail lets an auditor reconstruct exactly how many records entered and left the queue — the DOT/FMCSA rule mapping reference resolves any gap into the completeness field a compliance reviewer reads.

Verification

The tests use pytest-asyncio and prove the two properties that matter: the producer actually blocks when the queue is full (backpressure is real, not aspirational), and no records are lost across a burst.

import pytest


@pytest.mark.asyncio
async def test_producer_blocks_when_queue_is_full():
    queue = make_bounded_queue(maxsize=2)
    await queue.put(TelemetryRecord("bin-A17", 1, "2026-07-16T09:00:00+00:00", 41.0))
    await queue.put(TelemetryRecord("bin-A17", 2, "2026-07-16T09:00:05+00:00", 42.0))

    blocked = asyncio.create_task(
        queue.put(TelemetryRecord("bin-A17", 3, "2026-07-16T09:00:10+00:00", 43.0))
    )
    await asyncio.sleep(0.05)            # give the loop a chance to run the put
    assert not blocked.done()           # the third put is suspended: backpressure holds

    await queue.get()                    # free one slot
    await asyncio.sleep(0.05)
    assert blocked.done()                # the freed slot releases the producer


@pytest.mark.asyncio
async def test_no_records_lost_across_a_burst():
    committed = await run_ingestion(maxsize=8, batch_size=4)
    assert len(committed) == 50
    seqs = sorted(r.seq for r in committed)
    assert seqs == list(range(50))       # every record, exactly once, none dropped

Common Errors

The worker’s RSS climbs until the OOM killer sends SIGKILL. Root cause: the queue was created with asyncio.Queue() or asyncio.Queue(maxsize=0), so it is unbounded and a fast producer buffers the entire burst in memory. Fix: always pass a positive maxsize (Step 2 rejects zero outright) so await queue.put(...) can suspend the producer and convert the slow consumer into a slow producer instead of unbounded memory.

asyncio.QueueFull raised from the producer. Root cause: the producer used queue.put_nowait(record), which never waits — it raises the instant the queue is full. put_nowait is fire-and-forget and defeats the entire point of a bounded queue. Fix: use await queue.put(record) on the ingestion path (Step 3) so a full queue throttles the producer rather than raising; reserve put_nowait for places where dropping is an acceptable, explicitly logged decision.

ValueError: task_done() called too many times or queue.join() hangs forever. Root cause: task_done() was not called exactly once per put(), so the queue’s unfinished-task counter is wrong. Fix: pair every consumed item with exactly one task_done(), count the end-of-stream sentinels the same way, and — as Step 5 does — prefer awaiting the consumer task directly over queue.join() when the sentinel already signals completion.

FAQ

Why block the producer instead of dropping the oldest record?

Dropping trades a memory problem for a data-loss problem. On a compliance ingestion path a discarded fill-level reading is a missed service record that municipal reporting cannot account for, so dropped must stay zero. Blocking the producer pushes the slowdown back to the network layer, which can itself apply backpressure to the sensor’s send window — the pressure propagates safely instead of manifesting as silent loss.

How do I size maxsize and batch_size?

Size maxsize to the largest burst you want to absorb in memory without touching disk — a few thousand small records is usually a few megabytes, which is safe. Size batch_size to the commit cost of your writer: large enough that per-commit overhead amortizes, small enough that the tail flushed on shutdown is cheap. When in doubt keep maxsize a small multiple of batch_size so the queue smooths jitter without hoarding.

Up: Async Batch Processing

Related

• Async Batch Processing — the parent layer that fans validated fixes out to the routing solver under flow control.
• Validating GPS Coordinates in Python — the kinematic pre-filter whose accepted fixes feed this queue.
• Polling GPS Telematics Without Rate Limiting — the upstream reader whose send window this queue’s backpressure ultimately throttles.
• VRP Route Optimization Algorithms — the solver stage that consumes the batches this queue commits.