Tuning Backpressure and Semaphores for Async Ingest

An asyncio meter-read ingest that runs fine all month can fall over on the one night it matters. Month-end pulls the whole service territory at once, and the naive pattern — asyncio.gather over every read — spawns hundreds of thousands of coroutines that each want a database connection and each hold a parsed record in memory. The pool has fifty connections; the heap is not infinite. The result is TimeoutError on connection acquisition, an OOM kill, or a slow-motion collapse where retries pile on top of an already-saturated pool. The fix is not a bigger machine; it is bounding the work with a bounded queue and an asyncio.Semaphore sized to the downstream pool, so producers slow down when consumers fall behind. This page sits under async batch processing for high-volume reads, part of the broader Meter Data Ingestion & Validation Pipelines subsystem, and shows how to keep concurrency and memory flat through the peak.

# Minimal Prerequisites

• Python 3.11+ — for asyncio.TaskGroup, structured cancellation, and zoneinfo in the standard library.
• An async database driver with a bounded connection pool — the pool’s maximum size is the single most important number in this design; the semaphore is sized to it.
• decimal.Decimal for every metered quantity — concurrency changes throughput, never numeric representation; reads stay decimal-safe end to end.
• zoneinfo (not pytz) — reads carry timezone-aware timestamps so per-meter ordering is compared on true instants, not wall-clock strings.

Data assumptions: reads arrive from an upstream source (a file, an export endpoint, a topic) faster than the database can absorb them, each carries a stable meter_id and a monotonically increasing seq, and correctness requires that reads for the same meter are written in sequence order. Reads for different meters may be written concurrently.

# Annotated Implementation

The shape is a classic producer/consumer with two bounds working together. The bounded queue caps how many parsed records live in memory at once — when it fills, await queue.put() blocks the producer, which is backpressure. The semaphore caps how many writes hit the database concurrently, sized to the connection pool so the pool is never oversubscribed. Per-meter ordering is preserved by routing each meter to a stable worker rather than letting any worker grab any read.

from __future__ import annotations

import asyncio
from dataclasses import dataclass
from decimal import Decimal

# --- Tunables: the semaphore is sized to the pool, the queue to memory ---
DB_POOL_SIZE = 20                       # hard ceiling from the async driver
CONCURRENCY = DB_POOL_SIZE - 2          # leave headroom for health checks / retries
QUEUE_MAXSIZE = CONCURRENCY * 8         # bounded buffer → backpressure when full
NUM_WORKERS = 8                         # stable shards for per-meter ordering


@dataclass(frozen=True)
class Read:
    meter_id: str
    seq: int
    register: Decimal                   # never float


async def producer(source, queue: asyncio.Queue[Read]) -> None:
    """Feed reads in; block on a full queue instead of buffering unboundedly."""
    async for raw in source:
        # If consumers fall behind, put() blocks here — the producer feels backpressure
        # and stops pulling from the source, so memory stays flat.
        await queue.put(Read(**raw))
    for _ in range(NUM_WORKERS):
        await queue.put(None)           # one sentinel per worker to drain cleanly


async def worker(
    queue: asyncio.Queue[Read],
    sem: asyncio.Semaphore,
    pool,
    last_seq: dict[str, int],
) -> None:
    """Consume reads; the semaphore gates concurrent DB writes to the pool size."""
    while True:
        read = await queue.get()
        try:
            if read is None:
                return                  # sentinel: this worker is done
            prev = last_seq.get(read.meter_id, 0)
            if read.seq <= prev:
                continue                # duplicate / late — already written, skip
            # Acquire before touching the pool: never more than CONCURRENCY writes at once.
            async with sem:
                async with pool.acquire() as conn:
                    await conn.execute(
                        "INSERT INTO reads(meter_id, seq, register) VALUES($1,$2,$3)"
                        " ON CONFLICT (meter_id, seq) DO NOTHING",
                        read.meter_id, read.seq, read.register,
                    )
            last_seq[read.meter_id] = read.seq
        finally:
            queue.task_done()


def shard_for(meter_id: str) -> int:
    # Stable hash → the same meter always lands on the same worker, so its reads
    # are dequeued and written in the order the producer enqueued them.
    return hash(meter_id) % NUM_WORKERS


async def run_ingest(source, pool) -> None:
    # One queue per shard keeps per-meter ordering without a global lock.
    queues = [asyncio.Queue[Read](maxsize=QUEUE_MAXSIZE) for _ in range(NUM_WORKERS)]
    sem = asyncio.Semaphore(CONCURRENCY)
    last_seq: dict[str, int] = {}

    async def fan_out() -> None:
        async for raw in source:
            read = Read(**raw)
            await queues[shard_for(read.meter_id)].put(read)   # backpressure per shard
        for q in queues:
            await q.put(None)

    async with asyncio.TaskGroup() as tg:
        tg.create_task(fan_out())
        for q in queues:
            tg.create_task(worker(q, sem, pool, last_seq))

Two numbers do the load-bearing work. CONCURRENCY is deliberately set below DB_POOL_SIZE, not equal to it — leaving a couple of connections free means a health check or a retry can still acquire one when the ingest is at full tilt, instead of deadlocking against itself. QUEUE_MAXSIZE bounds memory: with a full queue the producer blocks in put(), propagating backpressure all the way up to the source so the process never holds more than a fixed number of parsed records. Sharding by a stable hash of meter_id gives per-meter ordering for free — the same meter always enqueues on the same worker’s queue, so its reads are written in enqueue order — while different meters still write concurrently up to the semaphore ceiling. The ON CONFLICT DO NOTHING clause makes each write idempotent so a retried batch never double-inserts, the same contract the AMI/AMR feed synchronization layer relies on.

# Edge Cases and Billing Gotchas

• Unbounded queue OOM. Replacing the bounded asyncio.Queue with an unbounded one — or with a plain list the producer appends to — removes backpressure entirely. On a month-end run the producer races ahead and buffers the whole territory in memory until the process is OOM-killed. The maxsize is not a tuning nicety; it is the only thing that keeps memory flat when the source is faster than the database.
• Semaphore set too high exhausts the pool. If CONCURRENCY meets or exceeds DB_POOL_SIZE, workers acquire semaphore slots and then block waiting for a connection that never frees, and a health check or retry that also needs a connection deadlocks. Always size the semaphore below the pool and leave headroom, so acquiring a slot guarantees a connection is available.
• A slow consumer stalls the whole pipeline. One worker stuck on a slow query holds its semaphore slot, shrinking effective concurrency, while its shard’s queue fills and backpressures the producer. Put a timeout on each write and route persistent failures to a dead-letter path via error handling and retry workflows so one bad meter cannot freeze the run.
• Ordering broken by a shared worker pool. If any worker can pull any read, two reads for the same meter can be written out of sequence and corrupt the register delta. Sharding by a stable hash of meter_id binds each meter to one worker, preserving order per meter while different meters still write concurrently — the same ordering guarantee the reading anomaly detection algorithms downstream assume.

# Verification Snippet

Assert the two invariants that make the design safe under peak: concurrent DB writes never exceed the semaphore count, and reads for one meter are written in sequence order.

import asyncio
from decimal import Decimal


async def test_semaphore_caps_concurrent_writes() -> None:
    sem = asyncio.Semaphore(4)
    live = 0
    peak = 0

    async def fake_write() -> None:
        nonlocal live, peak
        async with sem:
            live += 1
            peak = max(peak, live)
            await asyncio.sleep(0.01)        # simulate a DB round-trip
            live -= 1

    await asyncio.gather(*(fake_write() for _ in range(50)))
    assert peak <= 4                          # never oversubscribed the "pool"


def test_shard_preserves_per_meter_order() -> None:
    # Every read for a meter routes to one worker, so enqueue order == write order.
    reads = [("MTR-1", 1), ("MTR-1", 2), ("MTR-1", 3)]
    shards = {shard_for(m) for m, _ in reads}
    assert len(shards) == 1                    # one meter → exactly one worker
    seqs = [s for _, s in reads]
    assert seqs == sorted(seqs)                # sequence order is monotonic

The first test proves the semaphore is a real ceiling: fifty simultaneous writers never put more than four in flight, so a pool of that size is never oversubscribed. The second proves a single meter’s reads all land on one worker, which is what turns FIFO queue order into correct per-meter write order.

# FAQ

# Related Topics

• Async Batch Processing for High-Volume Reads — the parent guide on the decoupled asynchronous architecture this tuning sits inside.
• Meter Data Ingestion & Validation Pipelines — the end-to-end subsystem this ingest feeds.
• AMI/AMR Feed Synchronization Protocols — the idempotent, sequence-keyed handoff the workers write against.
• Error Handling & Retry Workflows — timeouts, retries, and dead-letter routing for writes that stall or fail.
• Reading Anomaly Detection Algorithms — statistical checks that assume the per-meter ordering this design preserves.

Up: Async Batch Processing for High-Volume Reads · Meter Data Ingestion & Validation Pipelines