Asyncio vs Celery Worker Pools for Invoice Batch Processing

When a carrier drops a hundred thousand invoices at once, you have to choose a concurrency model — an in-process asyncio pool or a broker-backed Celery pool — and picking the wrong one either starves your event loop or drowns tiny tasks in queue overhead.

The Failure You Are Hitting

Both models “work” on a laptop with a hundred sample invoices. The failure shows up only at real batch size, and it looks different depending on which model you guessed wrong:

• You wrote an asyncio.gather() over the whole batch and it runs beautifully until a subset of invoices needs a CPU-bound step — a Decimal-heavy reconciliation or a PDF re-parse — and those synchronous calls block the event loop. Throughput collapses to single-threaded, latency on the I/O tasks spikes, and the health check times out because the loop never yields.
• You put every invoice on a Celery queue including trivial ones that do nothing but a 5 ms schema check, and now each task pays broker round-trip, serialization, and result-backend write overhead that dwarfs the work. The batch that should finish in seconds takes twenty minutes, and Redis memory balloons with result payloads nobody reads.
• The batch half-completes and dies. With naive gather(), one unhandled exception cancels the sibling coroutines and you lose the in-flight work with no record of where it stopped. With Celery misconfigured for acks_late=False, a worker that crashes mid-task silently drops the message and those invoices are never processed — and nobody notices until reconciliation comes up short.

This decision lives inside Async Batch Processing Workflows, the tier that fans large carrier drops out across workers so the synchronous parsers upstream never have to. The model you choose determines whether that fan-out has back-pressure and durability or just optimism.

Root Cause Analysis

The mismatch is not a bug in asyncio or Celery. It comes from four properties of the workload that a small sample never exercises:

1. I/O-bound vs CPU-bound is not uniform across the batch. Most invoice processing is I/O — contract-store lookups, ledger writes, HTTP calls to a rate service. A minority of invoices trip a CPU-heavy path. Asyncio excels at the first and is actively harmful for the second, because a single blocking call freezes every concurrent coroutine sharing the loop.
2. Back-pressure has to exist somewhere. An unbounded gather() schedules every coroutine at once, so a 100k-invoice batch opens 100k concurrent connections to the rate store and knocks it over. Without a semaphore or a broker’s prefetch limit, concurrency equals batch size — which is not concurrency control, it is a denial-of-service on your own dependencies.
3. Durability is a property of where the work lives. In-process asyncio holds the work queue in memory; a pod restart loses everything in flight. A broker persists the queue, so a crashed worker’s un-acked messages redeliver. The choice is really about whether losing in-flight work on restart is acceptable.
4. Observability granularity differs. Asyncio gives you one process to watch; per-task state is whatever you log. Celery gives you queue depth, per-task retries, and worker heartbeats out of the box — at the cost of a broker and a result backend to operate.

The resolution is to stop treating it as one-or-the-other for the whole system. Use a bounded asyncio pool for the I/O-bound majority, hand CPU-bound and durability-critical work to Celery, and encode an explicit rule for which path a task takes.

The dimensions that decide the model:

Reproducible Diagnostic

Do not guess whether your batch is I/O-bound or CPU-bound — measure where wall-clock time actually goes. This snippet times the two phases of processing one representative invoice, which is the fact that decides the model:

import time, asyncio

def cpu_step(record: dict) -> dict:
    # Stand-in for Decimal reconciliation / re-parse: pure CPU, blocks the loop.
    total = 0
    for line in record["lines"]:
        total += int(round(float(line["freight"]) * 1000))
    record["checksum"] = total
    return record

async def io_step(record: dict) -> dict:
    await asyncio.sleep(0.02)          # stand-in for a contract-store lookup
    record["resolved"] = True
    return record

async def profile(record: dict, n: int = 500) -> None:
    t0 = time.perf_counter()
    for _ in range(n):
        cpu_step(dict(record))
    cpu = time.perf_counter() - t0

    t0 = time.perf_counter()
    await asyncio.gather(*(io_step(dict(record)) for _ in range(n)))
    io = time.perf_counter() - t0

    print(f"cpu_total={cpu:.3f}s io_wall={io:.3f}s ratio_cpu={cpu / (cpu + io):.2%}")

if __name__ == "__main__":
    sample = {"lines": [{"freight": "123.45"}, {"freight": "67.80"}]}
    asyncio.run(profile(sample))

Read the ratio like a decision table:

If the CPU step is meaningful and you keep it in asyncio, io_wall will inflate as n grows because the CPU loops block the scheduler — that inflation is the event-loop starvation you are trying to avoid, visible before you ship it.

Resolution Path

Build a bounded asyncio pool for the I/O majority, a Celery task for the CPU-and-durability minority, and a single decision rule that routes each invoice. Pin the dependencies:

# requirements.txt
celery==5.4.0
redis==5.2.1
structlog==24.4.0

Step 1 — Bound the asyncio pool so concurrency never equals batch size

Never gather() a raw batch. A semaphore caps in-flight coroutines, and return_exceptions=True keeps one poison invoice from cancelling its siblings:

# pools/async_pool.py
import asyncio
from typing import Awaitable, Callable

async def run_async_pool(
    invoices: list[dict],
    handler: Callable[[dict], Awaitable[dict]],
    max_concurrency: int = 32,
) -> list[dict | BaseException]:
    """Bounded fan-out for I/O-bound invoice work with built-in back-pressure."""
    sem = asyncio.Semaphore(max_concurrency)

    async def _guarded(inv: dict) -> dict:
        async with sem:                    # blocks past the limit — back-pressure
            return await handler(inv)

    # One poison invoice returns as an exception instead of killing the batch.
    return await asyncio.gather(*(_guarded(i) for i in invoices), return_exceptions=True)

Common mistake: setting max_concurrency to the batch size “to go faster.” The limit exists to protect the rate store; sizing it to the batch removes the back-pressure entirely and reproduces the DoS-on-your-own-dependency failure.

Step 2 — Retry the I/O path in-loop with bounded backoff

Asyncio has no built-in retry, so a transient lookup failure must be handled explicitly or it becomes a lost record:

# pools/retry.py
import asyncio, structlog

logger = structlog.get_logger()

async def with_retry(coro_factory, attempts: int = 3, base_delay: float = 0.25):
    """Retry a coroutine factory with exponential backoff; give up after `attempts`."""
    for attempt in range(1, attempts + 1):
        try:
            return await coro_factory()
        except (ConnectionError, TimeoutError) as exc:
            if attempt == attempts:
                logger.error("io_retry_exhausted", attempt=attempt, err=str(exc))
                raise
            await asyncio.sleep(base_delay * 2 ** (attempt - 1))

Handle only transient errors here — a ValueError from a malformed invoice is not retryable and must surface immediately, not loop three times before failing.

Step 3 — Push CPU-bound and durability-critical work to Celery

CPU work belongs in worker processes so it runs in true parallel and cannot starve an event loop. acks_late plus reject_on_worker_lost guarantees a crashed worker’s task redelivers instead of vanishing:

# pools/celery_app.py
from celery import Celery

app = Celery("freight_batch", broker="redis://localhost:6379/0",
             backend="redis://localhost:6379/1")

app.conf.update(
    worker_prefetch_multiplier=1,     # one heavy task at a time — real back-pressure
    task_acks_late=True,              # ack only AFTER success, so crashes redeliver
    task_reject_on_worker_lost=True,  # requeue the message if the worker dies mid-task
    task_default_retry_delay=5,
    task_time_limit=120,
)

@app.task(bind=True, max_retries=3, autoretry_for=(ConnectionError,))
def reconcile_invoice(self, record: dict) -> dict:
    """CPU-bound reconciliation: runs in a worker process, not the event loop."""
    from decimal import Decimal
    total = sum(Decimal(str(line["freight"])) for line in record["lines"])
    record["reconciled_total"] = str(total)
    record["status"] = "RECONCILED"
    return record

The broader durability walkthrough — routing, result handling, and dead-lettering under load — is in Implementing Async Batch Invoice Processing with Celery.

Step 4 — Encode the decision rule that routes each invoice

The rule is what keeps the two pools from becoming a coin flip. Classify by whether the invoice trips the CPU path or needs durability, then dispatch:

# pools/router.py
from pools.async_pool import run_async_pool
from pools.celery_app import reconcile_invoice

def needs_celery(record: dict) -> bool:
    """Route to Celery when work is CPU-heavy or must survive a restart."""
    heavy = len(record.get("lines", [])) > 200        # large re-parse / reconciliation
    durable = record.get("amount_due_cents", 0) > 5_000_00   # high value: cannot silently drop
    return heavy or durable

async def dispatch_batch(invoices: list[dict], io_handler) -> dict:
    io_bound = [i for i in invoices if not needs_celery(i)]
    cpu_bound = [i for i in invoices if needs_celery(i)]
    io_results = await run_async_pool(io_bound, io_handler, max_concurrency=32)
    for inv in cpu_bound:
        reconcile_invoice.delay(inv)                  # persisted on the broker
    return {"io": len(io_results), "queued_to_celery": len(cpu_bound)}

Verification

Prove the back-pressure and durability guarantees hold, not just that the happy path runs. The load-bearing tests are that the semaphore actually caps concurrency and that the router sends heavy work to Celery:

import asyncio
from pools.async_pool import run_async_pool
from pools.router import needs_celery

def test_semaphore_caps_in_flight_concurrency():
    peak = 0
    live = 0
    lock = asyncio.Lock()

    async def handler(inv):
        nonlocal peak, live
        async with lock:
            live += 1
            peak = max(peak, live)
        await asyncio.sleep(0.01)
        async with lock:
            live -= 1
        return inv

    batch = [{"id": n} for n in range(500)]
    asyncio.run(run_async_pool(batch, handler, max_concurrency=16))
    assert peak <= 16          # never exceeded the limit despite 500 tasks

def test_heavy_invoice_routes_to_celery():
    assert needs_celery({"lines": [{}] * 300}) is True
    assert needs_celery({"lines": [{}] * 3, "amount_due_cents": 100}) is False
    assert needs_celery({"lines": [], "amount_due_cents": 900_00}) is True

def test_poison_invoice_does_not_cancel_batch():
    async def handler(inv):
        if inv["id"] == 7:
            raise ValueError("bad invoice")
        return inv
    out = asyncio.run(run_async_pool([{"id": n} for n in range(10)], handler))
    assert sum(isinstance(r, dict) for r in out) == 9   # nine survived, one isolated

In production the health signals are: a flat asyncio loop-lag metric (a rising lag means CPU work leaked onto the loop — reroute it to Celery), a Celery queue depth that drains rather than grows monotonically, and a redelivery count that is non-zero only around actual worker restarts. A queue that only grows means the CPU pool is undersized; loop lag that spikes means the classifier let a heavy task onto the wrong path.

Preventive Configuration

Encode the routing thresholds and pool limits as configuration so a tuning change is a config edit, not a redeploy of logic:

batch_processing:
  asyncio:
    max_concurrency: 32          # cap in-flight lookups; back-pressure on the rate store
    retry_attempts: 3
    retry_base_delay_s: 0.25
    retryable_errors: [ConnectionError, TimeoutError]   # never retry ValueError
  celery:
    worker_prefetch_multiplier: 1   # one heavy task per worker — no hoarding
    task_acks_late: true            # ack after success so crashes redeliver
    reject_on_worker_lost: true
    max_retries: 3
    task_time_limit_s: 120
  router:
    celery_line_threshold: 200      # invoices heavier than this go to processes
    celery_value_threshold_cents: 500000   # high-value invoices need durability

• Assert the semaphore cap in CI. The concurrency test above runs on every build, so a refactor that drops the limit and lets gather() run unbounded fails immediately.
• Alert on loop lag. Export event-loop lag and alert above a small ceiling — the earliest signal that CPU work is starving the I/O pool and belongs on Celery.
• Never acks_late=False for money work. A worker crash with early ack silently drops the invoice; keep late acks and reject_on_worker_lost so high-value work always redelivers.
• Bound the queue depth alert. A Celery queue that grows for two consecutive intervals means the process pool is undersized — scale workers, do not raise prefetch, which only defeats the back-pressure.

FAQ

Can I just use asyncio for everything and skip running a broker?

Only if your work is genuinely all I/O-bound and losing in-flight tasks on a pod restart is acceptable. The moment a subset of invoices needs a CPU-heavy step, those synchronous calls block the shared event loop and collapse throughput for every concurrent task. Measure the CPU ratio with the diagnostic first; if it is meaningful, or if high-value invoices must survive a crash, you need the durability and process parallelism Celery provides.

Why is my Celery batch of tiny invoices slower than a plain loop?

Because each task pays broker round-trip, serialization, and result-backend overhead that dwarfs a 5 ms schema check. Celery earns its keep on CPU-heavy or durability-critical work, not on trivial I/O tasks. Route the small I/O-bound majority through a bounded asyncio pool and reserve Celery for the tasks whose per-item work is large enough to amortize the queue overhead.

How does each model apply back-pressure so I don't overwhelm the rate store?

In asyncio, an asyncio.Semaphore caps the number of in-flight coroutines, so concurrency is bounded no matter how large the batch. In Celery, worker_prefetch_multiplier=1 plus a fixed worker count bounds how many messages are processed at once. Both give you a hard ceiling on concurrent load against downstream dependencies; an unbounded gather() has none and is effectively a self-inflicted denial of service.

What guarantees an invoice is not lost if a worker crashes mid-batch?

Only the broker-backed path guarantees it. Set task_acks_late=True and task_reject_on_worker_lost=True so a worker acknowledges a message only after the task succeeds; if it crashes first, the un-acked message redelivers to another worker. In-process asyncio holds the queue in memory, so a restart loses everything in flight — which is exactly why durability-critical invoices route to Celery.

Related

• Async Batch Processing Workflows — the parent tier that fans carrier drops out across workers.
• Implementing Async Batch Invoice Processing with Celery — the durable broker-backed path in depth, with routing and dead-lettering.
• Ingestion Transport Protocols — how batches arrive over SFTP, AS2, and webhooks before this fan-out.
• Automated Invoice Parsing & EDI/XML Ingestion — the ingestion architecture this concurrency layer serves.
• EDI 210/810 Processing — the synchronous parser whose output these pools fan out at scale.

Up one level: Async Batch Processing Workflows · Section: Automated Invoice Parsing & EDI/XML Ingestion