Bounded Concurrency for Analyzer Result Batches with asyncio

Problem Statement

When a chemistry line finishes a rack, a hundred OBX results can land on the broker inside one second, and the naive reflex — asyncio.gather(*[commit(r) for r in batch]) — opens a hundred database connections at once, exhausts the LIMS connection pool, and turns a throughput problem into an outage that stalls every other analyzer feeding the same instance. The failure is not that the code is slow; it is that it has no ceiling. Unbounded concurrency trades a bounded queue you can reason about for an unbounded fan-out you cannot, and under a STAT surge the first symptom is TimeoutError on connection acquisition, followed by results that never commit because the event loop is saturated pushing work that the downstream cannot absorb. This page implements the concurrency ceiling as an explicit, testable control: an asyncio.Semaphore that caps in-flight commits, a TaskGroup that supervises the workers, and a bounded queue that propagates backpressure to the consumer so the broker — not the process heap — holds the surplus. It is the bounded-concurrency detail underneath the Async Batch Processing stage contract.

Prerequisites

Before wiring a concurrency ceiling into the drain loop, confirm the baseline:

  • Runtime: Python 3.11+ for asyncio.TaskGroup and asyncio.timeout, plus pydantic>=2.6 for the result model and pytest>=8 with pytest-asyncio for the fixtures.
  • Durable source: payloads already captured onto a broker with an immutable tracking_id by the Serial / FTP Polling Architectures layer; this stage consumes acknowledged messages, it does not poll instruments.
  • A real connection pool: an async database or HTTP client whose maximum size is a known, finite number (for example an asyncpg pool with max_size, or an httpx.AsyncClient with an explicit limits). The whole point of the semaphore is to keep concurrent work at or below that number.
  • Idempotent commit: a write that is safe to retry, because a bounded pool still means a worker can be cancelled mid-flight. Replay safety is specified in the sibling page on idempotent batch replays for instrument result loads.

Step-by-Step Implementation

Step 1: Model the unit of work

Each queue item is one validated analyzer result. Model it with Pydantic v2 so a malformed payload cannot enter the worker pool; the concurrency layer should only ever see structurally sound work, having inherited that guarantee from schema validation upstream.

python
from __future__ import annotations

import asyncio
import datetime as dt

from pydantic import BaseModel, Field


class AnalyzerResult(BaseModel):
    tracking_id: str
    accession: str
    analyte_loinc: str
    instrument_id: str
    value: float
    unit: str
    resulted_at: dt.datetime


class CommitOutcome(BaseModel):
    tracking_id: str
    committed: bool
    attempts: int = Field(ge=1)
    error: str | None = None

Step 2: Cap in-flight work with a Semaphore

The asyncio.Semaphore is the ceiling. Size it to the connection pool, not to the batch: if the pool grants at most eight connections, a permit count of eight means the ninth result waits at the async with sem boundary instead of blocking on connection acquisition deep inside the driver, where the failure mode is an opaque timeout rather than orderly waiting.

python
class PoolExhausted(RuntimeError):
    """Raised when a commit cannot acquire a connection within its budget."""


async def commit_one(
    result: AnalyzerResult,
    sem: asyncio.Semaphore,
    writer: "ResultWriter",
) -> CommitOutcome:
    """Commit a single result, holding a permit for the whole DB round trip."""
    async with sem:  # permit released even if the body raises or is cancelled
        try:
            async with asyncio.timeout(5.0):
                await writer.upsert(result)
            return CommitOutcome(tracking_id=result.tracking_id, committed=True, attempts=1)
        except TimeoutError as exc:
            raise PoolExhausted(result.tracking_id) from exc

Holding the permit across the entire round trip — acquire connection, write, release — is deliberate. A permit that is released before the write completes lets more coroutines rush the pool than there are connections, which is the exhaustion you were trying to prevent.

Step 3: Supervise the workers with a TaskGroup

asyncio.TaskGroup (Python 3.11+) is the right supervisor because it has structured-cancellation semantics: if one commit raises a non-retryable error, the group cancels its siblings and surfaces the failure as an ExceptionGroup, rather than leaving orphaned tasks writing after the batch has logically failed. Bounded concurrency and structured supervision are complementary — the semaphore limits how many run at once; the task group defines what happens when one dies.

python
async def drain_batch(
    batch: list[AnalyzerResult],
    *,
    max_in_flight: int,
    writer: "ResultWriter",
) -> list[CommitOutcome]:
    sem = asyncio.Semaphore(max_in_flight)
    outcomes: list[CommitOutcome] = []

    async def worker(r: AnalyzerResult) -> None:
        outcomes.append(await commit_one(r, sem, writer))

    async with asyncio.TaskGroup() as tg:
        for r in batch:
            tg.create_task(worker(r))
    return outcomes

Every result in the batch gets a task, but only max_in_flight of them hold a permit at any instant; the rest are cheap suspended coroutines parked at the semaphore, costing a few kilobytes each rather than a connection apiece.

Step 4: Propagate backpressure with a bounded queue

Fanning an entire batch into tasks is fine for a rack of a hundred; it is not fine for a re-drive of fifty thousand buffered results, where creating fifty thousand task objects up front reintroduces an unbounded footprint. For a continuous drain, put a bounded asyncio.Queue between the broker consumer and a fixed set of long-lived workers. When the queue is full, await queue.put(...) blocks the consumer, which stops acknowledging broker messages, which leaves the surplus durably on the broker — backpressure all the way to the source.

python
async def run_pool(
    consume: "Consumer",
    writer: "ResultWriter",
    *,
    workers: int,
    queue_max: int,
) -> None:
    queue: asyncio.Queue[AnalyzerResult] = asyncio.Queue(maxsize=queue_max)
    sem = asyncio.Semaphore(workers)

    async def worker() -> None:
        while True:
            result = await queue.get()
            try:
                await commit_one(result, sem, writer)
            finally:
                queue.task_done()

    async with asyncio.TaskGroup() as tg:
        task_workers = [tg.create_task(worker()) for _ in range(workers)]
        async for result in consume:          # blocks on a full queue → backpressure
            await queue.put(result)
        await queue.join()                    # drain in-flight work
        for w in task_workers:
            w.cancel()

Here queue_max bounds memory and workers bounds concurrency independently: a deep queue smooths bursty arrivals, while a small worker count protects the pool. The two knobs let you tune latency and safety separately instead of conflating them into one gather.

Bounded-concurrency drain: broker to bounded queue to semaphore-gated workers to connection pool A left-to-right flow. A durable broker holding a burst of analyzer results feeds a consumer that puts work onto a bounded asyncio queue of fixed maxsize; when the queue is full the put blocks and backpressure propagates back to the broker, which retains the surplus. A fixed pool of worker coroutines pulls from the queue, and each must acquire one of N semaphore permits before committing. The semaphore permit count equals the size of the downstream connection pool, so concurrent commits never exceed the number of available connections and the pool is never exhausted. Committed results land in the LIMS database. Broker result burst durable consume Bounded queue maxsize = queue_max put() blocks when full backpressure Semaphore N permits N = pool max_size worker 1 worker 2 worker N acquire permit Connection pool max_size = N never exhausted LIMS database idempotent commit
The semaphore permit count equals the connection-pool size, so concurrent commits can never exceed available connections; a full bounded queue pushes backpressure to the broker.

Verification & Testing

The property worth proving is the ceiling: no matter how large the batch, the number of commits running simultaneously never exceeds max_in_flight. Assert it with a writer that records peak concurrency.

python
import asyncio
import pytest


class PeakWriter:
    def __init__(self) -> None:
        self.current = 0
        self.peak = 0

    async def upsert(self, result: AnalyzerResult) -> None:
        self.current += 1
        self.peak = max(self.peak, self.current)
        try:
            await asyncio.sleep(0.01)  # simulate a DB round trip
        finally:
            self.current -= 1


@pytest.mark.asyncio
async def test_concurrency_never_exceeds_ceiling():
    writer = PeakWriter()
    batch = [
        AnalyzerResult(tracking_id=f"T{i}", accession="A1", analyte_loinc="2823-3",
                       instrument_id="chem-1", value=4.1, unit="mmol/L",
                       resulted_at=dt.datetime.now(dt.timezone.utc))
        for i in range(200)
    ]
    outcomes = await drain_batch(batch, max_in_flight=8, writer=writer)
    assert len(outcomes) == 200
    assert writer.peak <= 8  # the ceiling held under a 200-result burst

Expected: the test passes and writer.peak settles at exactly 8 under load, proving the semaphore, not the batch size, governs concurrency. A second fixture should give one result a writer that raises a non-retryable error and assert that drain_batch surfaces an ExceptionGroup — confirming the TaskGroup cancels siblings rather than leaving orphaned commits in flight.

Compliance Note

Bounded concurrency is a patient-safety control, not only a performance tactic. CLIA §493.1291 requires that the laboratory’s information system release complete, accurate results, and a pool exhausted under surge is a mechanism by which results are dropped or delayed without anyone deciding to drop them. Capping in-flight commits and pushing surplus back to a durable broker means every acknowledged payload is either committed or still retained upstream — never lost to an out-of-memory event. Because a bounded worker can still be cancelled mid-write, the commit must be idempotent for the 21 CFR Part 11 expectation of an attributable, non-duplicated record to hold; pair this ceiling with the replay-safe upsert specified for idempotent loads. Record CommitOutcome.attempts into the append-only audit sink so the retry history behind any released result is reconstructable on inspection.

Troubleshooting

Commits start failing with connection-acquisition timeouts under a STAT surge.

Root cause: the semaphore permit count is larger than the connection pool, so more coroutines reach the driver than there are connections and the surplus times out acquiring one. Fix: set the Semaphore count equal to (or below) the pool max_size, and hold the permit across the whole round trip in commit_one so a permit maps one-to-one to a connection in use.

Memory climbs steadily during a large re-drive.

Root cause: fanning the entire re-drive into TaskGroup.create_task up front allocates one task per result, so fifty thousand results become fifty thousand live objects. Fix: switch from drain_batch to run_pool with a bounded asyncio.Queue; a fixed worker count pulls from the queue and await queue.put(...) blocks the consumer when the queue is full, capping memory regardless of backlog size.

One bad result silently kills the whole batch and later results never commit.

Root cause: using bare asyncio.gather without return_exceptions, so the first raised exception propagates and the remaining tasks are abandoned in an undefined state. Fix: supervise with asyncio.TaskGroup, which cancels siblings deterministically and raises an ExceptionGroup you can route — retryable failures back to the queue, non-retryable ones to a dead-letter quarantine — instead of losing them.

Increasing the worker count made throughput worse, not better.

Root cause: worker count exceeds the useful concurrency of the downstream, so added workers only contend for the same connections and add scheduling overhead. Fix: tune workers to the pool size and use queue_max — not more workers — to absorb bursts; deepen the queue to smooth arrival spikes while keeping concurrency pinned at the pool ceiling.

Shutdown drops in-flight results.

Root cause: cancelling workers before the queue is drained discards items still pending task_done. Fix: on shutdown call await queue.join() to let in-flight commits finish before cancelling the worker tasks, and rely on idempotent upserts so any commit interrupted by a hard kill is safely replayed on restart.

Part of: Async Batch Processing.